fix: retry POP3/IMAP, DNS 8.8.8.8 fallback, debug counter, notification backoff

Agent-Logs-Url: https://github.com/christianlouis/InboxConverge/sessions/5ad49738-4b0b-4ed7-8686-07adb0ba9e5d

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>

backend/alembic/versions/0003_add_debug_counter_and_notif_flag.py

@@ -0,0 +1,45 @@
+"""Add debug_logging_run_count and error_notification_sent columns
+
+Revision ID: 0003
+Revises: 0002
+Create Date: 2026-05-03
+
+Changes:
+- debug_logging_run_count (INTEGER, default 0): counts completed/partial_failure
+  runs since debug_logging was last enabled.  Auto-disables debug_logging once
+  this counter reaches 5.  The counter is reset to 0 whenever debug_logging is
+  toggled back on via the API.
+
+- error_notification_sent (BOOLEAN, default FALSE): tracks whether a failure
+  notification has already been dispatched for the current consecutive error
+  streak.  Prevents notification spam: only the first failure in a streak fires
+  a notification.  Cleared (and a recovery notice sent) when a run succeeds.
+"""
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "0003"
+down_revision = "0002"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        "ALTER TABLE mail_accounts "
+        "ADD COLUMN IF NOT EXISTS debug_logging_run_count INTEGER NOT NULL DEFAULT 0"
+    )
+    op.execute(
+        "ALTER TABLE mail_accounts "
+        "ADD COLUMN IF NOT EXISTS error_notification_sent BOOLEAN NOT NULL DEFAULT FALSE"
+    )
+
+
+def downgrade() -> None:
+    op.execute(
+        "ALTER TABLE mail_accounts DROP COLUMN IF EXISTS debug_logging_run_count"
+    )
+    op.execute(
+        "ALTER TABLE mail_accounts DROP COLUMN IF EXISTS error_notification_sent"
+    )

backend/app/api/v1/endpoints/mail_accounts.py

@@ -180,6 +180,11 @@ async def update_mail_account(
         if password:  # Only update when a non-empty password is provided
             update_data["encrypted_password"] = encrypt_credential(password)
 
+    # When debug logging is (re-)enabled, reset the run counter so the user
+    # always gets exactly 5 debugged runs from the moment they check the box.
+    if update_data.get("debug_logging") is True:
+        update_data["debug_logging_run_count"] = 0
+
     for field, value in update_data.items():
         setattr(account, field, value)
 

backend/app/models/database_models.py

@@ -173,6 +173,15 @@ class MailAccount(Base):
 
     # Debug logging
     debug_logging = Column(Boolean, default=False)
+    # Number of completed/partial_failure runs since debug_logging was last
+    # enabled.  Resets to 0 when debug_logging is toggled True via the API.
+    # debug_logging is auto-disabled once this reaches 5.
+    debug_logging_run_count = Column(Integer, default=0)
+
+    # Notification backoff: True once an error notification has been sent for
+    # the current consecutive failure streak.  Cleared (with a recovery notice)
+    # when a run succeeds so the next new failure streak triggers a fresh alert.
+    error_notification_sent = Column(Boolean, default=False)
 
     # Statistics
     total_emails_processed = Column(Integer, default=0)

backend/app/services/mail_processor.py

@@ -9,6 +9,7 @@ import re
 import smtplib
 import socket
 import ssl
+import struct
 import threading
 import time as _time
 from datetime import datetime, timezone
@@ -44,19 +45,109 @@ def _set_cached_ipv4(host: str, port: int, ipv4: str) -> None:
         _dns_cache[(host, port)] = ipv4
 
 
+# ---------------------------------------------------------------------------
+# DNS fallback via Google 8.8.8.8
+# ---------------------------------------------------------------------------
+# When both the system resolver and the local cache fail we send a raw DNS
+# A-record query directly to 8.8.8.8:53.  This requires no extra dependencies
+# and keeps mail flowing even when /etc/resolv.conf points at a temporarily
+# unreachable resolver.
+
+_GOOGLE_DNS = "8.8.8.8"
+_DNS_PORT = 53
+_DNS_TIMEOUT = 3.0  # seconds
+
+
+def _build_dns_query(hostname: str) -> bytes:
+    """Build a minimal DNS A-record query packet for *hostname*."""
+    txid = 0xAB12  # arbitrary fixed transaction ID – we parse by TID match
+    flags = 0x0100  # standard query, recursion desired
+    header = struct.pack(">HHHHHH", txid, flags, 1, 0, 0, 0)
+    # Encode hostname as DNS labels: each label is length + bytes, terminated by 0x00
+    qname = b""
+    for label in hostname.rstrip(".").split("."):
+        encoded = label.encode()
+        qname += struct.pack("B", len(encoded)) + encoded
+    qname += b"\x00"
+    question = qname + struct.pack(">HH", 1, 1)  # QTYPE=A, QCLASS=IN
+    return header + question
+
+
+def _parse_dns_a_response(data: bytes, hostname: str) -> Optional[str]:
+    """Extract the first A-record IPv4 address from a raw DNS response packet."""
+    if len(data) < 12:
+        return None
+    txid, flags, qdcount, ancount = struct.unpack(">HHHH", data[:8])
+    if txid != 0xAB12 or ancount == 0:
+        return None
+    # Skip the header (12 bytes) and question section.
+    pos = 12
+    # Skip QDCOUNT questions: each is QNAME + QTYPE(2) + QCLASS(2)
+    for _ in range(qdcount):
+        while pos < len(data) and data[pos] != 0:
+            if data[pos] & 0xC0 == 0xC0:  # compression pointer
+                pos += 2
+                break
+            pos += 1 + data[pos]
+        else:
+            pos += 1  # null terminator
+        pos += 4  # skip QTYPE + QCLASS
+    # Parse answer records
+    for _ in range(ancount):
+        if pos >= len(data):
+            break
+        # Skip NAME field (may be a compression pointer or a label sequence)
+        if data[pos] & 0xC0 == 0xC0:
+            pos += 2
+        else:
+            while pos < len(data) and data[pos] != 0:
+                pos += 1 + data[pos]
+            pos += 1
+        if pos + 10 > len(data):
+            break
+        rtype, rclass, ttl, rdlength = struct.unpack(">HHIH", data[pos : pos + 10])
+        pos += 10
+        if rtype == 1 and rdlength == 4:  # A record
+            return "%d.%d.%d.%d" % tuple(data[pos : pos + 4])
+        pos += rdlength
+    return None
+
+
+def _query_google_dns_sync(hostname: str) -> Optional[str]:
+    """Blocking: send a DNS A-record query to 8.8.8.8 and return the IPv4 address."""
+    try:
+        query = _build_dns_query(hostname)
+        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        sock.settimeout(_DNS_TIMEOUT)
+        try:
+            sock.sendto(query, (_GOOGLE_DNS, _DNS_PORT))
+            data, _ = sock.recvfrom(512)
+        finally:
+            sock.close()
+        return _parse_dns_a_response(data, hostname)
+    except Exception:
+        return None
+
+
+async def _query_google_dns_async(hostname: str) -> Optional[str]:
+    """Async: run :func:`_query_google_dns_sync` in the default executor."""
+    loop = asyncio.get_running_loop()
+    try:
+        return await loop.run_in_executor(None, _query_google_dns_sync, hostname)
+    except Exception:
+        return None
+
+
 async def _resolve_ipv4(host: str, port: int) -> Optional[str]:
     """Async: resolve *host* to an IPv4 address using AF_INET getaddrinfo.
 
-    On success the result is stored in the module-level DNS cache so that a
-    subsequent call can return the cached address even if DNS is temporarily
-    unavailable.
+    Resolution order:
+    1. System resolver (AF_INET getaddrinfo) — result is cached on success.
+    2. Local DNS cache (last successful system-resolver result for this host).
+    3. Google public DNS (8.8.8.8) — used when both system resolver and cache
+       are unavailable.  Result is cached so subsequent calls benefit too.
 
-    When ``settings.DNS_CACHE_FALLBACK_ENABLED`` is True and live DNS fails
-    (e.g. EAI_AGAIN), the cached address from the last successful lookup is
-    returned instead, keeping connections alive through transient resolver
-    outages.
-
-    Returns None when no IPv4 address is available and the cache is empty.
+    Returns None only when all three strategies fail.
     """
     try:
         loop = asyncio.get_running_loop()
@@ -79,14 +170,25 @@ async def _resolve_ipv4(host: str, port: int) -> Optional[str]:
                     cached,
                 )
                 return cached
+        # Last resort: query 8.8.8.8 directly
+        google_ip = await _query_google_dns_async(host)
+        if google_ip:
+            logger.info(
+                "System DNS failed for %s:%s; resolved via 8.8.8.8 → %s",
+                host,
+                port,
+                google_ip,
+            )
+            if settings.DNS_CACHE_FALLBACK_ENABLED:
+                _set_cached_ipv4(host, port, google_ip)
+            return google_ip
     return None
 
 
 def _resolve_ipv4_sync(host: str, port: int) -> Optional[str]:
     """Sync counterpart of :func:`_resolve_ipv4` for use inside thread-pool executors.
 
-    Identical semantics: AF_INET lookup → cache on success, serve cache on
-    failure when ``settings.DNS_CACHE_FALLBACK_ENABLED`` is True.
+    Identical resolution order: system resolver → local cache → 8.8.8.8.
     """
     try:
         infos = socket.getaddrinfo(host, port, socket.AF_INET, socket.SOCK_STREAM)
@@ -106,6 +208,18 @@ def _resolve_ipv4_sync(host: str, port: int) -> Optional[str]:
                     cached,
                 )
                 return cached
+        # Last resort: query 8.8.8.8 directly
+        google_ip = _query_google_dns_sync(host)
+        if google_ip:
+            logger.info(
+                "System DNS failed for %s:%s; resolved via 8.8.8.8 → %s",
+                host,
+                port,
+                google_ip,
+            )
+            if settings.DNS_CACHE_FALLBACK_ENABLED:
+                _set_cached_ipv4(host, port, google_ip)
+            return google_ip
     return None
 
 
@@ -298,6 +412,41 @@ class MailForwardError(Exception):
     pass
 
 
+# ---------------------------------------------------------------------------
+# Transient-error detection (used by retry logic)
+# ---------------------------------------------------------------------------
+
+#: Maximum number of attempts for a POP3/IMAP connection.
+_MAX_FETCH_ATTEMPTS = 3
+#: Seconds to wait between retry attempts (fixed back-off).
+_FETCH_RETRY_DELAY = 5.0
+
+
+def _is_transient_error(exc: BaseException) -> bool:
+    """Return True for errors that are safe to retry (network glitches, EOF).
+
+    Authentication errors and protocol-level rejections should NOT be retried
+    as they will fail on every attempt regardless.
+    """
+    # EOF from POP3 server (connection dropped during or just after SSL handshake)
+    if isinstance(exc, poplib.error_proto):
+        msg = str(exc).lower()
+        return "eof" in msg or "timed out" in msg
+    # Network timeout (sync or async)
+    if isinstance(exc, (socket.timeout, asyncio.TimeoutError, TimeoutError)):
+        return True
+    # Server dropped the connection
+    if isinstance(exc, (ConnectionResetError, ConnectionAbortedError)):
+        return True
+    # IMAP server aborted the session
+    try:
+        if isinstance(exc, aioimaplib.Abort):
+            return True
+    except TypeError:
+        pass
+    return False
+
+
 def _format_connection_error(
     exc: BaseException,
     host: str = "",
@@ -669,9 +818,31 @@ class MailProcessor:
         seen: Set[str] = already_seen_uids or set()
 
         if self.account.protocol in [MailProtocol.POP3, MailProtocol.POP3_SSL]:
+            # POP3 already carries its own retry loop inside _fetch_pop3_emails.
             return await self._fetch_pop3_emails(effective_max, seen)
-        else:
-            return await self._fetch_imap_emails(effective_max, seen)
+
+        # IMAP: retry transient errors up to _MAX_FETCH_ATTEMPTS times.
+        last_exc: BaseException = RuntimeError("no attempts made")
+        for _attempt in range(1, _MAX_FETCH_ATTEMPTS + 1):
+            try:
+                return await self._fetch_imap_emails(effective_max, seen)
+            except MailFetchError as e:
+                last_exc = e
+                # Peek at the original cause to decide whether to retry.
+                cause = e.__cause__ or e
+                if _attempt < _MAX_FETCH_ATTEMPTS and _is_transient_error(cause):
+                    logger.warning(
+                        "Transient IMAP error on attempt %d/%d for account %s: %s; retrying in %.0f s",
+                        _attempt,
+                        _MAX_FETCH_ATTEMPTS,
+                        self.account.id,
+                        e,
+                        _FETCH_RETRY_DELAY,
+                    )
+                    await asyncio.sleep(_FETCH_RETRY_DELAY)
+                else:
+                    break
+        raise last_exc
 
     async def _fetch_pop3_emails(
         self, max_count: int, already_seen_uids: Set[str]
@@ -680,135 +851,149 @@ class MailProcessor:
         emails: List[bytes] = []
         new_uids: List[str] = []
 
-        try:
-            loop = asyncio.get_running_loop()
-            _dbg = self._debug  # capture for thread-pool closure
-            _host = str(self.account.host)
-            _port = int(self.account.port)
-            _protocol = self.account.protocol
+        loop = asyncio.get_running_loop()
+        _dbg = self._debug  # capture for thread-pool closure
+        _host = str(self.account.host)
+        _port = int(self.account.port)
+        _protocol = self.account.protocol
 
-            def fetch_pop3() -> Tuple[List[bytes], List[str]]:
-                _t_conn = _time.monotonic()
-                # Connect – prefer IPv4 to avoid ENETUNREACH on IPv6-only paths
-                _ipv4 = _resolve_ipv4_sync(_host, _port)
-                _ctx = (
-                    ssl.create_default_context()
-                    if _protocol == MailProtocol.POP3_SSL
-                    else None
-                )
-                pop_conn = _make_pop3_conn(_protocol, _host, _port, _ctx, 30, _ipv4)
-                if _dbg:
-                    _dbg.record(
-                        "connect",
-                        f"Connected to {_host}:{_port} "
-                        f"({'SSL' if _protocol == MailProtocol.POP3_SSL else 'plain'})"
-                        + (f" via {_ipv4}" if _ipv4 and _ipv4 != _host else ""),
-                        {"elapsed_ms": round((_time.monotonic() - _t_conn) * 1000)},
-                    )
-
-                # Authenticate
-                _t_auth = _time.monotonic()
-                pop_conn.user(str(self.account.username))
-                pop_conn.pass_(self.password)
-                if _dbg:
-                    _dbg.record(
-                        "auth",
-                        f"Authenticated as {self.account.username}",
-                        {"elapsed_ms": round((_time.monotonic() - _t_auth) * 1000)},
-                    )
-
-                # Retrieve UIDL map: {msg_number: uid_string}
-                uidl_response = pop_conn.uidl()
-                uid_map: Dict[int, str] = {}
-                for entry in uidl_response[1]:
-                    parts = entry.decode().split(" ", 1)
-                    if len(parts) == 2:
-                        uid_map[int(parts[0])] = parts[1].strip()
-
-                num_messages = len(uid_map)
-                if _dbg:
-                    _dbg.record(
-                        "uidl",
-                        f"Mailbox contains {num_messages} messages; "
-                        f"{len(already_seen_uids)} already downloaded",
-                        {"total": num_messages, "already_seen": len(already_seen_uids)},
-                    )
-                logger.info(
-                    f"Found {num_messages} messages for account {self.account.id}"
-                )
-
-                fetched: List[bytes] = []
-                fetched_uids: List[str] = []
-                fetched_count = 0
-
-                for msg_num, uid in uid_map.items():
-                    if fetched_count >= max_count:
-                        break
-
-                    # Skip messages we already processed
-                    if uid in already_seen_uids:
-                        logger.debug(
-                            f"Skipping already-downloaded message {uid} "
-                            f"for account {self.account.id}"
-                        )
-                        continue
-
-                    try:
-                        _t_msg = _time.monotonic()
-                        response, lines, octets = pop_conn.retr(msg_num)
-                        email_data = b"\r\n".join(lines)
-                        fetched.append(email_data)
-                        fetched_uids.append(uid)
-                        fetched_count += 1
-                        elapsed = round((_time.monotonic() - _t_msg) * 1000)
-                        logger.info(
-                            f"Retrieved message {msg_num} (uid={uid}) "
-                            f"from account {self.account.id}"
-                        )
-                        if _dbg:
-                            _dbg.record(
-                                "fetch_msg",
-                                f"Fetched message {msg_num} (uid={uid}): "
-                                f"{len(email_data)} bytes",
-                                {
-                                    "msg_num": msg_num,
-                                    "uid": uid,
-                                    "size_bytes": len(email_data),
-                                    "elapsed_ms": elapsed,
-                                },
-                            )
-                    except Exception as e:
-                        logger.error(f"Error retrieving message {msg_num}: {e}")
-                        if _dbg:
-                            _dbg.record(
-                                "fetch_error",
-                                f"Failed to retrieve message {msg_num} (uid={uid}): {e}",
-                                {"msg_num": msg_num, "uid": uid},
-                            )
-
-                pop_conn.quit()
-                if _dbg:
-                    _dbg.record(
-                        "quit",
-                        f"Disconnected — fetched {fetched_count} new message(s)",
-                        {"fetched": fetched_count},
-                    )
-                return fetched, fetched_uids
-
-            emails, new_uids = await loop.run_in_executor(None, fetch_pop3)
-
-        except Exception as e:
-            logger.error(f"Error fetching POP3 emails: {e}")
-            raise MailFetchError(
-                _format_connection_error(
-                    e,
-                    str(self.account.host),
-                    int(self.account.port),
-                    "POP3",
-                )
+        def fetch_pop3() -> Tuple[List[bytes], List[str]]:
+            _t_conn = _time.monotonic()
+            # Connect – prefer IPv4 to avoid ENETUNREACH on IPv6-only paths
+            _ipv4 = _resolve_ipv4_sync(_host, _port)
+            _ctx = (
+                ssl.create_default_context()
+                if _protocol == MailProtocol.POP3_SSL
+                else None
             )
+            pop_conn = _make_pop3_conn(_protocol, _host, _port, _ctx, 30, _ipv4)
+            if _dbg:
+                _dbg.record(
+                    "connect",
+                    f"Connected to {_host}:{_port} "
+                    f"({'SSL' if _protocol == MailProtocol.POP3_SSL else 'plain'})"
+                    + (f" via {_ipv4}" if _ipv4 and _ipv4 != _host else ""),
+                    {"elapsed_ms": round((_time.monotonic() - _t_conn) * 1000)},
+                )
 
-        return emails, new_uids
+            # Authenticate
+            _t_auth = _time.monotonic()
+            pop_conn.user(str(self.account.username))
+            pop_conn.pass_(self.password)
+            if _dbg:
+                _dbg.record(
+                    "auth",
+                    f"Authenticated as {self.account.username}",
+                    {"elapsed_ms": round((_time.monotonic() - _t_auth) * 1000)},
+                )
+
+            # Retrieve UIDL map: {msg_number: uid_string}
+            uidl_response = pop_conn.uidl()
+            uid_map: Dict[int, str] = {}
+            for entry in uidl_response[1]:
+                parts = entry.decode().split(" ", 1)
+                if len(parts) == 2:
+                    uid_map[int(parts[0])] = parts[1].strip()
+
+            num_messages = len(uid_map)
+            if _dbg:
+                _dbg.record(
+                    "uidl",
+                    f"Mailbox contains {num_messages} messages; "
+                    f"{len(already_seen_uids)} already downloaded",
+                    {"total": num_messages, "already_seen": len(already_seen_uids)},
+                )
+            logger.info(f"Found {num_messages} messages for account {self.account.id}")
+
+            fetched: List[bytes] = []
+            fetched_uids: List[str] = []
+            fetched_count = 0
+
+            for msg_num, uid in uid_map.items():
+                if fetched_count >= max_count:
+                    break
+
+                # Skip messages we already processed
+                if uid in already_seen_uids:
+                    logger.debug(
+                        f"Skipping already-downloaded message {uid} "
+                        f"for account {self.account.id}"
+                    )
+                    continue
+
+                try:
+                    _t_msg = _time.monotonic()
+                    response, lines, octets = pop_conn.retr(msg_num)
+                    email_data = b"\r\n".join(lines)
+                    fetched.append(email_data)
+                    fetched_uids.append(uid)
+                    fetched_count += 1
+                    elapsed = round((_time.monotonic() - _t_msg) * 1000)
+                    logger.info(
+                        f"Retrieved message {msg_num} (uid={uid}) "
+                        f"from account {self.account.id}"
+                    )
+                    if _dbg:
+                        _dbg.record(
+                            "fetch_msg",
+                            f"Fetched message {msg_num} (uid={uid}): "
+                            f"{len(email_data)} bytes",
+                            {
+                                "msg_num": msg_num,
+                                "uid": uid,
+                                "size_bytes": len(email_data),
+                                "elapsed_ms": elapsed,
+                            },
+                        )
+                except Exception as e:
+                    logger.error(f"Error retrieving message {msg_num}: {e}")
+                    if _dbg:
+                        _dbg.record(
+                            "fetch_error",
+                            f"Failed to retrieve message {msg_num} (uid={uid}): {e}",
+                            {"msg_num": msg_num, "uid": uid},
+                        )
+
+            pop_conn.quit()
+            if _dbg:
+                _dbg.record(
+                    "quit",
+                    f"Disconnected — fetched {fetched_count} new message(s)",
+                    {"fetched": fetched_count},
+                )
+            return fetched, fetched_uids
+
+        # Retry loop: transient errors (EOF, timeout, connection reset) are
+        # retried up to _MAX_FETCH_ATTEMPTS times with a short fixed delay.
+        last_exc: BaseException = RuntimeError("no attempts made")
+        for _attempt in range(1, _MAX_FETCH_ATTEMPTS + 1):
+            try:
+                emails, new_uids = await loop.run_in_executor(None, fetch_pop3)
+                return emails, new_uids
+            except Exception as e:
+                last_exc = e
+                if _attempt < _MAX_FETCH_ATTEMPTS and _is_transient_error(e):
+                    logger.warning(
+                        "Transient POP3 error on attempt %d/%d for account %s: %s; retrying in %.0f s",
+                        _attempt,
+                        _MAX_FETCH_ATTEMPTS,
+                        self.account.id,
+                        e,
+                        _FETCH_RETRY_DELAY,
+                    )
+                    await asyncio.sleep(_FETCH_RETRY_DELAY)
+                else:
+                    break
+
+        logger.error(f"Error fetching POP3 emails: {last_exc}")
+        raise MailFetchError(
+            _format_connection_error(
+                last_exc,
+                str(self.account.host),
+                int(self.account.port),
+                "POP3",
+            )
+        )
 
     async def _fetch_imap_emails(
         self, max_count: int, already_seen_uids: Set[str]
@@ -1077,7 +1262,7 @@ class MailProcessor:
                     int(self.account.port),
                     "IMAP",
                 )
-            )
+            ) from e
         finally:
             # Always attempt a clean logout.  If the server already sent BYE
             # the logout call will fail silently rather than masking the real

backend/app/workers/tasks.py

@@ -37,7 +37,7 @@ from app.services.gmail_service import GmailService, GmailAuthError
 from app.services.config_service import ConfigService
 from app.services.notification_service import send_user_notification
 from app.core.config import settings
-from sqlalchemy import select, delete, or_, func
+from sqlalchemy import select, delete, or_, update as sa_update
 
 logger = logging.getLogger(__name__)
 
@@ -538,30 +538,26 @@ async def process_mail_account(account_id: int):
             # Per-email failures are already tracked in ProcessingLog and the
             # run's emails_failed counter so the user can drill into them without
             # having the account badge stuck in ERROR indefinitely.
+            _was_in_error = account.status == AccountStatus.ERROR  # type: ignore[comparison-overlap]
+            _had_notified = bool(account.error_notification_sent)  # type: ignore[attr-defined]
             account.last_successful_check_at = datetime.now(timezone.utc)  # type: ignore[assignment]
             account.status = AccountStatus.ACTIVE  # type: ignore[assignment]
             account.last_error_message = None  # type: ignore[assignment]
             account.last_error_at = None  # type: ignore[assignment]
+            # Clear the notification-sent flag so a future error streak fires a fresh alert.
+            account.error_notification_sent = False  # type: ignore[assignment]
 
-            # Auto-disable debug logging after 5 completed runs in the past 24 h
-            # to prevent it being left on indefinitely.
+            # Auto-disable debug logging after 5 runs since it was last enabled.
+            # The counter resets to 0 each time the user turns the flag on via
+            # the API, so "5 runs" is always counted from the moment of enabling.
             if account.debug_logging:  # type: ignore[attr-defined]
-                _24h_ago = datetime.now(timezone.utc) - timedelta(hours=24)
-                _debug_run_count = (
-                    await db.execute(
-                        select(func.count(ProcessingRun.id)).where(
-                            ProcessingRun.mail_account_id == account.id,
-                            ProcessingRun.started_at >= _24h_ago,
-                            ProcessingRun.status.in_(["completed", "partial_failure"]),
-                        )
-                    )
-                ).scalar_one()
-                if _debug_run_count >= 5:
+                account.debug_logging_run_count = (account.debug_logging_run_count or 0) + 1  # type: ignore[attr-defined,assignment]
+                if account.debug_logging_run_count >= 5:  # type: ignore[attr-defined]
                     account.debug_logging = False  # type: ignore[assignment]
+                    account.debug_logging_run_count = 0  # type: ignore[assignment]
                     logger.info(
-                        "Auto-disabled debug logging for account %s after %d runs in 24 h",
+                        "Auto-disabled debug logging for account %s after 5 runs",
                         account.id,
-                        _debug_run_count,
                     )
 
             await db.commit()
@@ -569,9 +565,38 @@ async def process_mail_account(account_id: int):
             # Send failure notification after the commit so the status is
             # persisted even if the notification fails.  Use a fresh session
             # to avoid interfering with the (now-committed) main transaction.
-            if emails_failed > 0:
+            #
+            # Backoff policy:
+            #  - Recovery: if the account was previously in ERROR and we had
+            #    already sent a failure notification, send one recovery notice now.
+            #  - New error streak: only notify on the FIRST run that has failures
+            #    (error_notification_sent was False before this run).  Subsequent
+            #    failing runs stay silent until recovery resets the flag.
+            if _was_in_error and _had_notified:
                 try:
                     async with async_session_maker() as notif_db:
+                        await send_user_notification(
+                            db=notif_db,
+                            user_id=int(account.user_id),
+                            title="InboxRescue: Mail Account Recovered",
+                            body=f"Mail account '{account.name}' is processing normally again.",
+                            notify_on_error=True,
+                        )
+                except Exception as notify_exc:
+                    logger.warning(
+                        f"Failed to send recovery notification: {notify_exc}"
+                    )
+
+            if emails_failed > 0 and not _had_notified:
+                try:
+                    async with async_session_maker() as notif_db:
+                        # Persist the flag so the next failing run stays silent.
+                        async with notif_db.begin():
+                            await notif_db.execute(
+                                sa_update(MailAccount)
+                                .where(MailAccount.id == account.id)
+                                .values(error_notification_sent=True)
+                            )
                         await send_user_notification(
                             db=notif_db,
                             user_id=int(account.user_id),
@@ -655,18 +680,38 @@ async def process_mail_account(account_id: int):
             # to avoid the post-rollback session's broken greenlet context causing
             # the notification query itself to fail with "greenlet_spawn has not
             # been called".
+            #
+            # Backoff: only notify on the first failure in a consecutive error
+            # streak.  error_notification_sent is set True here and cleared on
+            # the next successful run, so repeat failures stay silent until the
+            # account recovers.
             if "account" in locals() and account is not None:
-                try:
-                    async with async_session_maker() as notif_db:
-                        await send_user_notification(
-                            db=notif_db,
-                            user_id=int(account.user_id),
-                            title="InboxRescue: Mail Processing Error",
-                            body=f"Error processing mail account '{account.name}': {e}",
-                            notify_on_error=True,
+                _already_notified = bool(
+                    getattr(account, "error_notification_sent", False)
+                )
+                if not _already_notified:
+                    try:
+                        async with async_session_maker() as notif_db:
+                            # Persist the flag first so even if the notification
+                            # delivery fails the flag is set and the next run won't
+                            # try again.
+                            async with notif_db.begin():
+                                await notif_db.execute(
+                                    sa_update(MailAccount)
+                                    .where(MailAccount.id == account.id)
+                                    .values(error_notification_sent=True)
+                                )
+                            await send_user_notification(
+                                db=notif_db,
+                                user_id=int(account.user_id),
+                                title="InboxRescue: Mail Processing Error",
+                                body=f"Error processing mail account '{account.name}': {e}",
+                                notify_on_error=True,
+                            )
+                    except Exception as notify_exc:
+                        logger.warning(
+                            f"Failed to send error notification: {notify_exc}"
                         )
-                except Exception as notify_exc:
-                    logger.warning(f"Failed to send error notification: {notify_exc}")
 
 
 @celery_app.task(base=AsyncTask, name="app.workers.tasks.process_all_enabled_accounts")

backend/tests/unit/test_dns_fallback.py

@@ -0,0 +1,214 @@
+"""
+Unit tests for the DNS 8.8.8.8 fallback resolver in mail_processor.py.
+
+Covers:
+  - _build_dns_query() produces a valid DNS A-record packet
+  - _parse_dns_a_response() extracts the first A record correctly
+  - _query_google_dns_sync() sends UDP query to 8.8.8.8 and returns an IPv4
+  - _resolve_ipv4_sync() falls through to 8.8.8.8 when system DNS and cache fail
+  - _resolve_ipv4_sync() caches the 8.8.8.8 result for subsequent calls
+"""
+
+import socket
+import struct
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from app.services.mail_processor import (
+    _build_dns_query,
+    _parse_dns_a_response,
+    _query_google_dns_sync,
+    _resolve_ipv4_sync,
+    _dns_cache,
+    _dns_cache_lock,
+)
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_dns_response(txid: int, ip: str) -> bytes:
+    """Craft a minimal valid DNS A-record response for the given IP."""
+    flags = 0x8180  # response, recursion available
+    qdcount = 1
+    ancount = 1
+    header = struct.pack(">HHHHHH", txid, flags, qdcount, ancount, 0, 0)
+    # Question: dummy single-label name "x" + QTYPE=A + QCLASS=IN
+    qname = b"\x01x\x00"
+    question = qname + struct.pack(">HH", 1, 1)
+    # Answer: pointer to question name (0xC00C), TYPE=A, CLASS=IN, TTL, RDLEN=4, IP
+    octets = tuple(int(o) for o in ip.split("."))
+    answer = struct.pack(">HHHiH", 0xC00C, 1, 1, 300, 4) + bytes(octets)
+    return header + question + answer
+
+
+# ---------------------------------------------------------------------------
+# _build_dns_query
+# ---------------------------------------------------------------------------
+
+
+class TestBuildDnsQuery:
+    def test_contains_hostname_labels(self):
+        data = _build_dns_query("pop.example.com")
+        # "pop" label should appear: length byte 3 followed by b"pop"
+        assert b"\x03pop" in data
+
+    def test_qtype_a_and_class_in(self):
+        data = _build_dns_query("mail.example.com")
+        # Last 4 bytes of question: QTYPE=0x0001, QCLASS=0x0001
+        assert data[-4:] == b"\x00\x01\x00\x01"
+
+    def test_transaction_id_is_0xAB12(self):
+        data = _build_dns_query("x.example.com")
+        txid = struct.unpack(">H", data[:2])[0]
+        assert txid == 0xAB12
+
+
+# ---------------------------------------------------------------------------
+# _parse_dns_a_response
+# ---------------------------------------------------------------------------
+
+
+class TestParseDnsAResponse:
+    def test_extracts_ip_from_valid_response(self):
+        response = _make_dns_response(0xAB12, "1.2.3.4")
+        result = _parse_dns_a_response(response, "example.com")
+        assert result == "1.2.3.4"
+
+    def test_returns_none_for_wrong_txid(self):
+        response = _make_dns_response(0x1234, "1.2.3.4")
+        result = _parse_dns_a_response(response, "example.com")
+        assert result is None
+
+    def test_returns_none_for_short_data(self):
+        assert _parse_dns_a_response(b"\x00\x01", "example.com") is None
+
+    def test_returns_none_when_no_answers(self):
+        # Build a header with ancount=0
+        header = struct.pack(">HHHHHH", 0xAB12, 0x8180, 0, 0, 0, 0)
+        assert _parse_dns_a_response(header, "x") is None
+
+
+# ---------------------------------------------------------------------------
+# _query_google_dns_sync
+# ---------------------------------------------------------------------------
+
+
+class TestQueryGoogleDnsSync:
+    def test_returns_ip_on_success(self):
+        response = _make_dns_response(0xAB12, "5.6.7.8")
+
+        mock_sock = MagicMock()
+        mock_sock.recvfrom.return_value = (response, ("8.8.8.8", 53))
+
+        with patch("socket.socket", return_value=mock_sock):
+            result = _query_google_dns_sync("pop.example.com")
+
+        assert result == "5.6.7.8"
+        mock_sock.sendto.assert_called_once()
+
+    def test_returns_none_on_socket_error(self):
+        with patch("socket.socket", side_effect=OSError("network down")):
+            result = _query_google_dns_sync("pop.example.com")
+        assert result is None
+
+    def test_returns_none_on_timeout(self):
+        mock_sock = MagicMock()
+        mock_sock.recvfrom.side_effect = socket.timeout("timed out")
+        with patch("socket.socket", return_value=mock_sock):
+            result = _query_google_dns_sync("pop.example.com")
+        assert result is None
+
+
+# ---------------------------------------------------------------------------
+# _resolve_ipv4_sync  — 8.8.8.8 fallback path
+# ---------------------------------------------------------------------------
+
+
+class TestResolveIpv4SyncGoogleFallback:
+    def setup_method(self):
+        """Clear the DNS cache before each test to avoid state bleed."""
+        with _dns_cache_lock:
+            _dns_cache.clear()
+
+    def test_falls_through_to_google_when_system_dns_and_cache_fail(self):
+        with (
+            patch(
+                "socket.getaddrinfo",
+                side_effect=OSError("Name or service not known"),
+            ),
+            patch(
+                "app.services.mail_processor._query_google_dns_sync",
+                return_value="9.10.11.12",
+            ) as mock_google,
+            patch(
+                "app.services.mail_processor.settings.DNS_CACHE_FALLBACK_ENABLED",
+                False,
+            ),
+        ):
+            result = _resolve_ipv4_sync("pop.web.de", 995)
+
+        assert result == "9.10.11.12"
+        mock_google.assert_called_once_with("pop.web.de")
+
+    def test_google_result_is_cached(self):
+        with (
+            patch(
+                "socket.getaddrinfo",
+                side_effect=OSError("Name or service not known"),
+            ),
+            patch(
+                "app.services.mail_processor._query_google_dns_sync",
+                return_value="9.10.11.12",
+            ),
+            patch(
+                "app.services.mail_processor.settings.DNS_CACHE_FALLBACK_ENABLED",
+                True,
+            ),
+        ):
+            _resolve_ipv4_sync("pop.web.de", 995)
+
+        with _dns_cache_lock:
+            cached = _dns_cache.get(("pop.web.de", 995))
+        assert cached == "9.10.11.12"
+
+    def test_returns_none_when_all_strategies_fail(self):
+        with (
+            patch(
+                "socket.getaddrinfo",
+                side_effect=OSError("Name or service not known"),
+            ),
+            patch(
+                "app.services.mail_processor._query_google_dns_sync",
+                return_value=None,
+            ),
+            patch(
+                "app.services.mail_processor.settings.DNS_CACHE_FALLBACK_ENABLED",
+                False,
+            ),
+        ):
+            result = _resolve_ipv4_sync("nonexistent.invalid", 995)
+
+        assert result is None
+
+    def test_system_dns_success_skips_google(self):
+        with (
+            patch(
+                "socket.getaddrinfo",
+                return_value=[(None, None, None, None, ("1.2.3.4", 995))],
+            ),
+            patch(
+                "app.services.mail_processor._query_google_dns_sync",
+            ) as mock_google,
+            patch(
+                "app.services.mail_processor.settings.DNS_CACHE_FALLBACK_ENABLED",
+                False,
+            ),
+        ):
+            result = _resolve_ipv4_sync("pop.example.com", 995)
+
+        assert result == "1.2.3.4"
+        mock_google.assert_not_called()

backend/tests/unit/test_mail_processor_pop3.py

@@ -854,3 +854,90 @@ class TestForwardEmail:
         body_text = body_part.get_payload(decode=True).decode("utf-8")
         # Body should have header info but no HTML content extracted
         assert "Originally from: x@y.com" in body_text
+
+
+# ---------------------------------------------------------------------------
+# Retry logic tests
+# ---------------------------------------------------------------------------
+
+
+class TestFetchPop3Retry:
+    """_fetch_pop3_emails retries on transient errors."""
+
+    @patch("app.services.mail_processor._resolve_ipv4_sync", return_value=None)
+    @patch("app.services.mail_processor.poplib")
+    @pytest.mark.asyncio
+    async def test_retries_on_eof_error(self, mock_poplib, _mock_resolve):
+        """EOF error on first attempt triggers a retry; second attempt succeeds."""
+        import asyncio
+        import poplib as real_poplib
+
+        # First call raises EOF; second succeeds
+        good_conn = MagicMock()
+        good_conn.stat.return_value = (0, 0)
+        good_conn.uidl.return_value = (b"+OK", [], 0)
+        good_conn.quit.return_value = None
+
+        mock_poplib.error_proto = real_poplib.error_proto
+        mock_poplib.POP3_SSL.side_effect = [
+            real_poplib.error_proto("-ERR EOF"),
+            good_conn,
+        ]
+
+        account = _make_account(protocol="pop3_ssl")
+        proc = MailProcessor(account, "secret")
+
+        with patch(
+            "app.services.mail_processor.asyncio.sleep", new_callable=AsyncMock
+        ) as mock_sleep:
+            emails, uids = await proc._fetch_pop3_emails(10, set())
+
+        assert emails == []
+        assert uids == []
+        # sleep should have been called once between attempt 1 and attempt 2
+        mock_sleep.assert_awaited_once()
+
+    @patch("app.services.mail_processor._resolve_ipv4_sync", return_value=None)
+    @patch("app.services.mail_processor.poplib")
+    @pytest.mark.asyncio
+    async def test_raises_after_max_attempts(self, mock_poplib, _mock_resolve):
+        """All attempts fail with EOF → MailFetchError is raised."""
+        import poplib as real_poplib
+
+        mock_poplib.error_proto = real_poplib.error_proto
+        mock_poplib.POP3_SSL.side_effect = real_poplib.error_proto("-ERR EOF")
+
+        account = _make_account(protocol="pop3_ssl")
+        proc = MailProcessor(account, "secret")
+
+        with patch(
+            "app.services.mail_processor.asyncio.sleep", new_callable=AsyncMock
+        ):
+            with pytest.raises(MailFetchError):
+                await proc._fetch_pop3_emails(10, set())
+
+    @patch("app.services.mail_processor._resolve_ipv4_sync", return_value=None)
+    @patch("app.services.mail_processor.poplib")
+    @pytest.mark.asyncio
+    async def test_no_retry_on_auth_error(self, mock_poplib, _mock_resolve):
+        """Authentication errors are not retried (non-transient)."""
+        import poplib as real_poplib
+
+        conn = MagicMock()
+        mock_poplib.error_proto = real_poplib.error_proto
+        mock_poplib.POP3_SSL.return_value = conn
+        # user() succeeds; pass_() raises auth error
+        conn.user.return_value = b"+OK"
+        conn.pass_.side_effect = real_poplib.error_proto("-ERR Authentication failed")
+
+        account = _make_account(protocol="pop3_ssl")
+        proc = MailProcessor(account, "secret")
+
+        with patch(
+            "app.services.mail_processor.asyncio.sleep", new_callable=AsyncMock
+        ) as mock_sleep:
+            with pytest.raises(MailFetchError):
+                await proc._fetch_pop3_emails(10, set())
+
+        # No sleep = no retry
+        mock_sleep.assert_not_awaited()

backend/tests/unit/test_tasks.py

@@ -1822,3 +1822,249 @@ class TestCleanupOldLogs:
 
             # Should not raise
             await cleanup_old_logs.run(days_to_keep=30)
+
+
+# ---------------------------------------------------------------------------
+# Debug-logging counter auto-disable tests
+# ---------------------------------------------------------------------------
+
+
+class TestDebugLoggingCounter:
+    """debug_logging is auto-disabled after exactly 5 runs since it was enabled."""
+
+    def _make_account_with_debug(self, run_count: int = 0) -> MagicMock:
+        account = _make_account(delivery_method=DeliveryMethod.GMAIL_API)
+        account.debug_logging = True
+        account.debug_logging_run_count = run_count
+        account.error_notification_sent = False
+        account.status = MagicMock(value="active")
+        return account
+
+    def _build_task_mocks(self, account):
+        maker, session = _mock_session_maker()
+        gmail_cred = _make_gmail_cred(user_id=account.user_id)
+
+        account_result = MagicMock()
+        account_result.scalar_one_or_none.return_value = account
+        seen_result = MagicMock()
+        seen_result.scalars.return_value.all.return_value = []
+        gmail_cred_result = MagicMock()
+        gmail_cred_result.scalar_one_or_none.return_value = gmail_cred
+        session.execute = AsyncMock(
+            side_effect=[account_result, seen_result, gmail_cred_result]
+        )
+        session.commit = AsyncMock()
+        session.refresh = AsyncMock()
+        return maker, session, gmail_cred
+
+    @pytest.mark.asyncio
+    async def test_counter_increments_each_run(self):
+        """debug_logging_run_count increments from 0 to 1 after one successful run."""
+        account = self._make_account_with_debug(run_count=0)
+        maker, session, gmail_cred = self._build_task_mocks(account)
+
+        mock_processor = AsyncMock()
+        mock_processor.fetch_emails.return_value = ([], [])
+        mock_gmail_svc = _make_gmail_service()
+
+        with (
+            patch(f"{MODULE}.async_session_maker", maker),
+            patch(f"{MODULE}.engine", AsyncMock()),
+            patch(f"{MODULE}.decrypt_credential", return_value="pw"),
+            patch(f"{MODULE}.MailProcessor", return_value=mock_processor),
+            patch(f"{MODULE}.GmailService", return_value=mock_gmail_svc),
+            patch(f"{MODULE}.send_user_notification", new_callable=AsyncMock),
+        ):
+            from app.workers.tasks import process_mail_account
+
+            await process_mail_account.run(1)
+
+        assert account.debug_logging_run_count == 1
+        assert account.debug_logging is True
+
+    @pytest.mark.asyncio
+    async def test_debug_disabled_at_run_5(self):
+        """debug_logging turns off and counter resets when it reaches 5."""
+        account = self._make_account_with_debug(run_count=4)
+        maker, session, gmail_cred = self._build_task_mocks(account)
+
+        mock_processor = AsyncMock()
+        mock_processor.fetch_emails.return_value = ([], [])
+        mock_gmail_svc = _make_gmail_service()
+
+        with (
+            patch(f"{MODULE}.async_session_maker", maker),
+            patch(f"{MODULE}.engine", AsyncMock()),
+            patch(f"{MODULE}.decrypt_credential", return_value="pw"),
+            patch(f"{MODULE}.MailProcessor", return_value=mock_processor),
+            patch(f"{MODULE}.GmailService", return_value=mock_gmail_svc),
+            patch(f"{MODULE}.send_user_notification", new_callable=AsyncMock),
+        ):
+            from app.workers.tasks import process_mail_account
+
+            await process_mail_account.run(1)
+
+        assert account.debug_logging is False
+        assert account.debug_logging_run_count == 0
+
+    @pytest.mark.asyncio
+    async def test_debug_not_disabled_at_run_4(self):
+        """debug_logging stays on at run 4 (need one more)."""
+        account = self._make_account_with_debug(run_count=3)
+        maker, session, gmail_cred = self._build_task_mocks(account)
+
+        mock_processor = AsyncMock()
+        mock_processor.fetch_emails.return_value = ([], [])
+        mock_gmail_svc = _make_gmail_service()
+
+        with (
+            patch(f"{MODULE}.async_session_maker", maker),
+            patch(f"{MODULE}.engine", AsyncMock()),
+            patch(f"{MODULE}.decrypt_credential", return_value="pw"),
+            patch(f"{MODULE}.MailProcessor", return_value=mock_processor),
+            patch(f"{MODULE}.GmailService", return_value=mock_gmail_svc),
+            patch(f"{MODULE}.send_user_notification", new_callable=AsyncMock),
+        ):
+            from app.workers.tasks import process_mail_account
+
+            await process_mail_account.run(1)
+
+        assert account.debug_logging is True
+        assert account.debug_logging_run_count == 4
+
+
+# ---------------------------------------------------------------------------
+# Notification backoff tests
+# ---------------------------------------------------------------------------
+
+
+class TestNotificationBackoff:
+    """Failure notifications should only fire once per error streak."""
+
+    def _make_smtp_account(self, **overrides):
+        account = _make_account(delivery_method=DeliveryMethod.SMTP, **overrides)
+        account.debug_logging = False
+        account.debug_logging_run_count = 0
+        account.error_notification_sent = overrides.get("error_notification_sent", False)
+        account.status = MagicMock(value="active")
+        return account
+
+    def _smtp_session(self, account):
+        maker, session = _mock_session_maker()
+        user_smtp = MagicMock()
+        user_smtp.host = "smtp.x.com"
+        user_smtp.port = 587
+        user_smtp.username = "u"
+        user_smtp.encrypted_password = "ep"
+        user_smtp.use_tls = True
+
+        account_result = MagicMock()
+        account_result.scalar_one_or_none.return_value = account
+        seen_result = MagicMock()
+        seen_result.scalars.return_value.all.return_value = []
+        smtp_result = MagicMock()
+        smtp_result.scalar_one_or_none.return_value = user_smtp
+        session.execute = AsyncMock(
+            side_effect=[account_result, seen_result, smtp_result]
+        )
+        session.commit = AsyncMock()
+        session.refresh = AsyncMock()
+        return maker, session
+
+    @pytest.mark.asyncio
+    async def test_first_failure_sends_notification(self):
+        """First email-forwarding failure in a streak sends a notification."""
+        raw_email = _build_raw_email()
+        account = self._make_smtp_account(error_notification_sent=False)
+        maker, session = self._smtp_session(account)
+
+        mock_processor = AsyncMock()
+        mock_processor.post_process_messages = AsyncMock()
+        mock_processor.fetch_emails.return_value = ([raw_email], ["uid-1"])
+
+        mock_notif_session = MagicMock()
+        mock_notif_ctx = MagicMock()
+        mock_notif_ctx.__aenter__ = AsyncMock(return_value=mock_notif_session)
+        mock_notif_ctx.__aexit__ = AsyncMock(return_value=False)
+        mock_notif_ctx.begin = MagicMock(return_value=mock_notif_ctx)
+        mock_notif_session.execute = AsyncMock()
+
+        def session_maker_side_effect():
+            return mock_notif_ctx
+
+        mock_send_notification = AsyncMock(return_value=1)
+
+        with (
+            patch(f"{MODULE}.async_session_maker", side_effect=[maker(), session_maker_side_effect()]),
+            patch(f"{MODULE}.engine", AsyncMock()),
+            patch(f"{MODULE}.decrypt_credential", return_value="pw"),
+            patch(f"{MODULE}.MailProcessor", return_value=mock_processor),
+            patch(
+                f"{MODULE}.MailProcessor.forward_email",
+                new_callable=AsyncMock,
+                return_value=False,
+            ),
+            patch(f"{MODULE}.send_user_notification", mock_send_notification),
+        ):
+            from app.workers.tasks import process_mail_account
+
+            await process_mail_account.run(1)
+
+        mock_send_notification.assert_awaited()
+
+    @pytest.mark.asyncio
+    async def test_second_failure_suppressed(self):
+        """When error_notification_sent is already True, no new notification fires."""
+        raw_email = _build_raw_email()
+        account = self._make_smtp_account(error_notification_sent=True)
+        maker, session = self._smtp_session(account)
+
+        mock_processor = AsyncMock()
+        mock_processor.post_process_messages = AsyncMock()
+        mock_processor.fetch_emails.return_value = ([raw_email], ["uid-1"])
+
+        mock_send_notification = AsyncMock(return_value=0)
+
+        with (
+            patch(f"{MODULE}.async_session_maker", maker),
+            patch(f"{MODULE}.engine", AsyncMock()),
+            patch(f"{MODULE}.decrypt_credential", return_value="pw"),
+            patch(f"{MODULE}.MailProcessor", return_value=mock_processor),
+            patch(
+                f"{MODULE}.MailProcessor.forward_email",
+                new_callable=AsyncMock,
+                return_value=False,
+            ),
+            patch(f"{MODULE}.send_user_notification", mock_send_notification),
+        ):
+            from app.workers.tasks import process_mail_account
+
+            await process_mail_account.run(1)
+
+        mock_send_notification.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_recovery_clears_flag(self):
+        """A successful run clears error_notification_sent."""
+        account = self._make_smtp_account(error_notification_sent=False)
+        # Put account in "active" status – recovery notice fires when was ERROR
+        account.status = MagicMock(value="active")
+        maker, session = self._smtp_session(account)
+
+        mock_processor = AsyncMock()
+        mock_processor.fetch_emails.return_value = ([], [])
+        mock_processor.post_process_messages = AsyncMock()
+
+        with (
+            patch(f"{MODULE}.async_session_maker", maker),
+            patch(f"{MODULE}.engine", AsyncMock()),
+            patch(f"{MODULE}.decrypt_credential", return_value="pw"),
+            patch(f"{MODULE}.MailProcessor", return_value=mock_processor),
+            patch(f"{MODULE}.send_user_notification", new_callable=AsyncMock),
+        ):
+            from app.workers.tasks import process_mail_account
+
+            await process_mail_account.run(1)
+
+        # Flag is reset to False after successful connection
+        assert account.error_notification_sent is False