From a622db320580febb0d9573d73f128cbaeacc6504 Mon Sep 17 00:00:00 2001 From: duffyduck Date: Tue, 29 Sep 2026 13:47:50 +0200 Subject: [PATCH] reconnect on dropped IMAP connection instead of failing the whole batch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The server dropped the connection mid-poll once ("Broken pipe"); the scheduler kept using the dead connection, so the remaining ~400 mails of the batch each failed and were still marked as processed — i.e. silently never evaluated again. - Fetch retries once after reconnecting on connection-level errors (IMAP4.abort, OSError incl. timeouts/SSL). If the reconnect fails, the poll stops with a single warning, keeps the progress so far and leaves the remaining mails unmarked for the next poll. - A failed action followed by a dead connection triggers a reconnect; that mail is not marked as processed, so it is retried next poll. - IMAPClient.reconnect() / is_alive() helpers. Checked with a simulated server: single drop (all mails processed, no errors), permanent outage (one warning, rest unmarked), drop during a move (mail left unmarked for retry). Co-Authored-By: Claude Opus 5.5 --- app/services/imap_client.py | 20 ++++++++++++ app/services/scheduler.py | 64 +++++++++++++++++++++++++++++++++---- 2 files changed, 78 insertions(+), 6 deletions(-) diff --git a/app/services/imap_client.py b/app/services/imap_client.py index e8d819a..2af84d8 100644 --- a/app/services/imap_client.py +++ b/app/services/imap_client.py @@ -23,6 +23,10 @@ imaplib._MAXLINE = 100_000_000 # Maximale Wartezeit pro IMAP-Socket-Operation (connect, Lesen, Schreiben). IMAP_TIMEOUT_SECONDS = 120 +# Fehler, die eine getrennte/kaputte Verbindung bedeuten (Broken pipe, EOF, +# Timeout, SSL-Fehler) — im Gegensatz zu Fehlern einer einzelnen Mail. +CONNECTION_ERRORS = (imaplib.IMAP4.abort, OSError) + def _fetch_payload(msg_data) -> bytes | None: """Sucht in einer FETCH-Antwort den Teil mit den Mail-Bytes. Die Antwort kann @@ -190,6 +194,22 @@ class IMAPClient: self._conn.login(self.username, self.password) logger.info("IMAP verbunden: %s@%s", self.username, self.host) + def reconnect(self) -> None: + """Verbindung neu aufbauen, z.B. nachdem der Server sie getrennt hat.""" + if self._conn: + try: + self._conn.shutdown() + except Exception: + pass + self._conn = None + self.connect() + + def is_alive(self) -> bool: + try: + return self.conn.noop()[0] == "OK" + except Exception: + return False + def disconnect(self) -> None: if self._conn: try: diff --git a/app/services/scheduler.py b/app/services/scheduler.py index e130a72..a86c692 100644 --- a/app/services/scheduler.py +++ b/app/services/scheduler.py @@ -10,7 +10,7 @@ from app.database import SessionLocal from app.models.db_models import Account, FilterRule, LogLevel, ProcessedMail from app.services.encryption import decrypt from app.services.filter_engine import evaluate_conditions, execute_action -from app.services.imap_client import IMAPClient +from app.services.imap_client import CONNECTION_ERRORS, IMAPClient from app.services.log_service import write_log from app.services.maintenance import run_maintenance @@ -23,6 +23,35 @@ scheduler = AsyncIOScheduler() COMMIT_EVERY = 25 +class ConnectionLost(Exception): + """Die IMAP-Verbindung ist weg und ließ sich nicht wiederherstellen.""" + + +def _reconnect(client: IMAPClient, reason: Exception | str) -> None: + logger.warning("IMAP-Verbindung unterbrochen (%s) — verbinde neu", reason) + client._current_folder = None + try: + client.reconnect() + except Exception as e: + raise ConnectionLost(f"{reason}; Neuverbindung fehlgeschlagen: {e}") from e + + +def _fetch_with_reconnect(client: IMAPClient, folder: str, uid: str): + """Mail holen; bei Verbindungsabbruch einmal neu verbinden und wiederholen. + Ohne das scheiterten nach einem einzigen Abbruch alle restlichen Mails des + Polls ('Broken pipe') und wurden trotzdem als verarbeitet markiert.""" + for attempt in (1, 2): + try: + if getattr(client, "_current_folder", None) != folder: + client.select_folder(folder) + client._current_folder = folder + return client.fetch_mail(uid) + except CONNECTION_ERRORS as e: + if attempt == 2: + raise ConnectionLost(str(e)) from e + _reconnect(client, e) + + def _verbose_log(**kwargs) -> None: """Routine-Einträge (Nicht-Übereinstimmung, 'keine neuen Mails', Poll-Start) erzeugten Millionen Logzeilen ohne Nutzen — nur bei LOG_VERBOSE schreiben.""" @@ -178,11 +207,9 @@ def _poll_account_sync(account_id: int) -> None: mail = mail_cache[cache_key] else: try: - # Ordner muss ausgewählt sein - if not hasattr(client, '_current_folder') or client._current_folder != folder: - client.select_folder(folder) - client._current_folder = folder - mail = client.fetch_mail(uid) + mail = _fetch_with_reconnect(client, folder, uid) + except ConnectionLost: + raise except Exception as e: write_log( message=f"Fehler beim Abrufen von Mail {uid}", @@ -218,6 +245,7 @@ def _poll_account_sync(account_id: int) -> None: rule.match_count = (rule.match_count or 0) + 1 rule.last_match_at = datetime.utcnow() # Aktionen ausführen + retry_later = False for action in rule.actions: success = execute_action( client, mail, action, smtp_config, @@ -247,6 +275,13 @@ def _poll_account_sync(account_id: int) -> None: else: total_errors += 1 reason = client.last_error or "unbekannter Grund" + if not client.is_alive(): + # Verbindung weg: neu verbinden und die Mail beim + # nächsten Poll erneut versuchen statt sie als + # verarbeitet abzuhaken + _reconnect(client, reason) + retry_later = True + reason += " (Verbindung unterbrochen — wird beim nächsten Poll erneut versucht)" write_log( message=f"Aktion fehlgeschlagen: {action_label} — {reason}"[:1000], level=LogLevel.ERROR, @@ -276,6 +311,9 @@ def _poll_account_sync(account_id: int) -> None: db=db, ) + if matched and retry_later: + continue + # Mail für DIESE Regel als verarbeitet markieren db.add(ProcessedMail( account_id=account.id, @@ -314,6 +352,20 @@ def _poll_account_sync(account_id: int) -> None: account.last_poll_at = datetime.utcnow() db.commit() + except ConnectionLost as e: + # Bisherigen Fortschritt behalten; die restlichen Mails sind nicht als + # verarbeitet markiert und werden beim nächsten Poll geprüft. + logger.error("IMAP-Verbindung für Konto %s verloren: %s", account_id, e) + try: + db.commit() + except Exception: + db.rollback() + write_log( + message="Verbindung zum Mailserver unterbrochen — Poll abgebrochen, restliche Mails werden beim nächsten Poll geprüft", + level=LogLevel.WARNING, + account_id=account_id, + details=str(e), + ) except Exception as e: logger.error("Fehler beim Polling von Konto %s: %s", account_id, e) db.rollback()