diff --git a/CHANGELOG.md b/CHANGELOG.md index 82eca1d..4163df0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,33 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 +## [Unreleased] + +### Fixed + +- **POP3/IMAP connectivity**: Added retry logic (up to 3 attempts, 5 s delay) for + transient errors (`-ERR EOF`, timeout, connection-reset) in both POP3 and IMAP + fetch paths. Most `-ERR EOF` and "timed out" failures now self-heal without + surfacing as errors. + +- **DNS fallback to 8.8.8.8**: When both the system resolver and the in-process + DNS cache fail (e.g. transient resolver outage), mail-processor now sends a raw + UDP A-record query directly to Google's public resolver (8.8.8.8:53). The + result is cached in the same way as system-resolver results. No new runtime + dependencies. + +- **Debug-logging auto-disable**: Counter now counts 5 runs *from when the checkbox + was last ticked*, not in a rolling 24-hour window. New `debug_logging_run_count` + column (migration `0003`) is reset to 0 each time `debug_logging` is enabled via + the API, so users always get exactly 5 debugged runs per enable. + +- **Notification backoff**: Error notifications (both connection-level failures and + per-email forwarding failures) are now sent only once per consecutive failure + streak. A `error_notification_sent` flag (migration `0003`) tracks whether the + current streak has already been notified. When the account recovers the flag is + cleared and a single recovery notification is sent; the next failure streak will + then fire a fresh alert. + ## v0.10.1 (2026-05-03) ### Bug Fixes diff --git a/backend/alembic/versions/0003_add_debug_counter_and_notif_flag.py b/backend/alembic/versions/0003_add_debug_counter_and_notif_flag.py new file mode 100644 index 0000000..085ab1b --- /dev/null +++ b/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" + ) diff --git a/backend/app/api/v1/endpoints/mail_accounts.py b/backend/app/api/v1/endpoints/mail_accounts.py index a738c27..2416fbb 100644 --- a/backend/app/api/v1/endpoints/mail_accounts.py +++ b/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) diff --git a/backend/app/models/database_models.py b/backend/app/models/database_models.py index 9a4926e..016e431 100644 --- a/backend/app/models/database_models.py +++ b/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) diff --git a/backend/app/services/mail_processor.py b/backend/app/services/mail_processor.py index 451e429..6c97d5c 100644 --- a/backend/app/services/mail_processor.py +++ b/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 diff --git a/backend/app/workers/tasks.py b/backend/app/workers/tasks.py index c9a5434..6d9144a 100644 --- a/backend/app/workers/tasks.py +++ b/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") diff --git a/backend/tests/unit/test_dns_fallback.py b/backend/tests/unit/test_dns_fallback.py new file mode 100644 index 0000000..06bacf2 --- /dev/null +++ b/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() diff --git a/backend/tests/unit/test_mail_processor_pop3.py b/backend/tests/unit/test_mail_processor_pop3.py index b7f27e9..da9566f 100644 --- a/backend/tests/unit/test_mail_processor_pop3.py +++ b/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() diff --git a/backend/tests/unit/test_tasks.py b/backend/tests/unit/test_tasks.py index 71d602b..f0d0ec2 100644 --- a/backend/tests/unit/test_tasks.py +++ b/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 diff --git a/docs/TODO.md b/docs/TODO.md index 4bb60db..b7719cd 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -34,8 +34,21 @@ Comprehensive task breakdown for repository improvements and production readines `debug_logging` boolean column (migration `0002`), `MailDebugRecorder` class, instrumented all connection phases (connect, auth, select, search, fetch UIDs, per-message fetch, logout), persisted as `ProcessingLog[level=DEBUG]`. - Auto-disables after 5 runs in 24 h. Toggle in account edit form. Connection - trace viewer in Mailbox Activity logs page. + Auto-disables after 5 runs **from when the flag was last enabled** (migration + `0003` adds `debug_logging_run_count` counter; the old 24-hour window is + replaced). Toggle in account edit form. Connection trace viewer in Mailbox + Activity logs page. + +- [x] **POP3/IMAP retry + DNS 8.8.8.8 fallback**: Transient errors (`-ERR EOF`, + timeout, connection-reset) are retried up to 3 times (5 s fixed delay) in both + POP3 and IMAP paths. DNS resolution now falls back to a raw UDP query to + 8.8.8.8:53 when both the system resolver and the in-process cache fail, keeping + connections alive even if the container's resolver is temporarily unavailable. + +- [x] **Notification backoff**: Error notifications fire only once per consecutive + failure streak (`error_notification_sent` flag, migration `0003`). A recovery + notice is sent when the account processes successfully again, then the flag + resets so the next failure streak triggers a fresh alert. - [x] **Google OAuth consent screen legal compliance**: Added English Privacy Policy (`/privacy`) with Google API Limited Use Disclosure, Terms of Service (`/terms`), legal footer links on the home page (resolves Google's "homepage has no privacy policy link" verification rejection), login page, and register page (consent text). Cross-link from `/datenschutz` to `/privacy` added.