Unreachable mail servers: send retries in Scheduled, real action retries, account health

Sending: a send the server could not take (unreachable, timeout, 4xx) is no
longer failed at once. It waits in Scheduled ("retrying") and is tried again
after 1, 2, 5, 10, 30 min, then hourly, for 24 hours — on the server, so it
works with the browser closed. A permanent refusal (5xx, wrong password) is
"failed" at once, with the server's reason. GET /outbox lists both;
POST /outbox/{id}/retry (Retry now) and /cancel (Edit, Delete) act on them,
and they are left out of Drafts. Mail delivered but not filed in Sent now
counts as sent, and a conditional claim keeps two jobs from sending one
message twice. The app says when a send failed and when it finally went.

Actions: arq retries a job only on Retry, never on a plain exception, so the
"retry x5, then revert" for flags and moves never happened. Flag, move,
archive, restore, delete and empty-folder jobs now raise Retry while the
server is unreachable (~24 hours of growing gaps), skip a change a later one
replaced, stop at the first connection failure instead of 20 s per message,
and delete/empty keep the rows until the server has the messages gone.

Account health: 3 failed connections in a row mark an account unreachable
(amber in the sidebar and Settings, cleared on the next success); 3 refused
sign-ins stop its syncing as auth_failed until the password is entered
again, so the provider does not lock the account. Sync loops back off 30 s
doubling to 10 min and log one line per expected failure; a refused
connection fails at once with its reason (the connect task is kept), and a
silently dead IDLE connection is noticed within ~11 min instead of ~29.

Also: the empty reading pane is clipped to its width when dragged very
narrow; tests no longer leave live accounts and outbox rows in the database.

Migration 0021: outbox attempts / next_attempt_at / retry_until, accounts
sync_error / sync_error_since.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-05 22:08:42 +02:00
co-authored by Claude Opus 5.5
parent e1bda37188
commit 609e7c58f9
31 changed files with 1375 additions and 151 deletions
+3
View File
@@ -101,6 +101,9 @@ async def list_accounts(user: CurrentUser, db: DbSession) -> list[AccountOut]:
sync_window=a.sync_window,
backfill_total=a.backfill_total,
backfill_done=a.backfill_done,
sync_error=a.sync_error,
sync_error_since=a.sync_error_since.isoformat() if a.sync_error_since else None,
state_detail=a.state_detail,
)
for a in accounts
]
+76 -2
View File
@@ -16,10 +16,12 @@ from app.config import get_settings
from app.db import get_session
from app.identity.deps import CurrentUser, require_csrf
from app.indexer import blobstore
from app.models import Attachment, BodyPart, ComposeAttachment, Draft, Mailbox, Message
from app.models import Attachment, BodyPart, ComposeAttachment, Draft, Mailbox, Message, Outbox
from app.outbox.jobs import RETRY_WINDOW
from app.outbox.schemas import (
DraftIn,
DraftOut,
OutboxItemOut,
RemoteDraftOut,
SendRequest,
SendResponse,
@@ -29,6 +31,7 @@ from app.outbox.service import (
cancel_outbox,
create_outbox,
delete_mirrored_draft_row,
not_being_sent,
upsert_draft,
)
from app.queue import enqueue_job
@@ -109,7 +112,9 @@ def _draft_out(draft: Draft) -> DraftOut:
@router.get("/drafts", response_model=list[DraftOut])
async def list_drafts(user: CurrentUser, db: DbSession) -> list[DraftOut]:
drafts = await db.scalars(
select(Draft).where(Draft.user_id == user.id).order_by(Draft.updated_at.desc())
select(Draft)
.where(Draft.user_id == user.id, not_being_sent())
.order_by(Draft.updated_at.desc())
)
return [_draft_out(d) for d in drafts]
@@ -313,3 +318,72 @@ async def undo_send(outbox_id: uuid.UUID, user: CurrentUser, db: DbSession) -> d
if not ok:
raise_api_error(409, "undo_too_late", "the message has already left the undo window")
return {"status": "cancelled"}
# --- the Scheduled folder's server side: sends being retried, or failed ------------
@router.get("/outbox", response_model=list[OutboxItemOut])
async def list_outbox(user: CurrentUser, db: DbSession) -> list[OutboxItemOut]:
"""Sends that did not go out: retrying (the server could not take them yet)
or failed for good. They wait in Scheduled, not in Drafts."""
rows = await db.execute(
select(Outbox, Draft)
.join(Draft, Draft.id == Outbox.draft_id)
.where(Outbox.user_id == user.id, Outbox.state.in_(("retrying", "failed")))
.order_by(Outbox.send_after.desc())
)
out: list[OutboxItemOut] = []
for ob, draft in rows.tuples():
payload = draft.payload or {}
to = payload.get("to")
out.append(
OutboxItemOut(
outbox_id=ob.id,
draft_id=draft.id,
account_id=draft.account_id,
state=ob.state,
attempts=ob.attempts,
error=ob.error,
next_attempt_at=ob.next_attempt_at.isoformat() if ob.next_attempt_at else None,
retry_until=ob.retry_until.isoformat() if ob.retry_until else None,
sent_at=ob.send_after.isoformat(),
to=[str(t) for t in to] if isinstance(to, list) else [],
subject=str(payload.get("subject") or ""),
preview=str(payload.get("text") or "")[:200],
)
)
return out
async def _waiting_send(db: AsyncSession, user_id: uuid.UUID, outbox_id: uuid.UUID) -> Outbox:
ob = await db.scalar(select(Outbox).where(Outbox.id == outbox_id, Outbox.user_id == user_id))
if ob is None or ob.state not in ("retrying", "failed"):
raise_api_error(409, "not_waiting", "this message is not waiting to be sent")
return ob
@router.post("/outbox/{outbox_id}/retry")
async def retry_outbox(outbox_id: uuid.UUID, user: CurrentUser, db: DbSession) -> dict[str, str]:
"""Retry now. A send that had failed for good gets a fresh 24-hour window."""
ob = await _waiting_send(db, user.id, outbox_id)
now = datetime.now(UTC)
if ob.state == "failed" or ob.retry_until is None or ob.retry_until <= now:
ob.retry_until = now + RETRY_WINDOW
ob.state = "retrying"
ob.next_attempt_at = now
await db.commit()
await enqueue_job("send_job", str(ob.id), _job_id=f"send:{ob.id}:now:{ob.attempts}")
return {"status": "retrying"}
@router.post("/outbox/{outbox_id}/cancel")
async def cancel_waiting_send(
outbox_id: uuid.UUID, user: CurrentUser, db: DbSession
) -> dict[str, str]:
"""Stop trying: the message goes back to Drafts (Edit, or before Delete)."""
ob = await _waiting_send(db, user.id, outbox_id)
ob.state = "cancelled"
ob.next_attempt_at = None
await db.commit()
return {"status": "cancelled", "draft_id": str(ob.draft_id)}
+4 -1
View File
@@ -13,6 +13,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.db import get_session
from app.identity.deps import CurrentUser
from app.models import Draft, Mailbox, Message
from app.outbox.service import not_being_sent
router = APIRouter(prefix="/mailboxes", tags=["mail"])
@@ -58,7 +59,9 @@ async def folder_counts(user: CurrentUser, db: DbSession) -> FolderCounts:
# Drafts folder = local drafts + IMAP-Drafts messages we haven't adopted (mirror
# UIDs excluded), matching what list_drafts + list_remote_drafts return.
local = await db.scalar(select(func.count()).select_from(Draft).where(Draft.user_id == user.id))
local = await db.scalar(
select(func.count()).select_from(Draft).where(Draft.user_id == user.id, not_being_sent())
)
owned = {
(r.account_id, r.imap_uid)
for r in await db.execute(
+5
View File
@@ -97,6 +97,11 @@ class AccountOut(BaseModel):
sync_window: str = "all"
backfill_total: int = 0
backfill_done: int = 0
# "unreachable" while the mail server has not answered for a while.
sync_error: str | None = None
sync_error_since: str | None = None
# Why the state is what it is (e.g. the server's reply to a refused sign-in).
state_detail: str = ""
class SearchThreadHit(BaseModel):
+7
View File
@@ -137,6 +137,9 @@ class Account(Base):
backfill_total: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0"))
backfill_done: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0"))
last_sync_at: Mapped[dt.datetime | None] = mapped_column(TIMESTAMP(timezone=True))
# The server has not answered for a while ("unreachable"); cleared on success.
sync_error: Mapped[str | None] = mapped_column(Text)
sync_error_since: Mapped[dt.datetime | None] = mapped_column(TIMESTAMP(timezone=True))
created_at: Mapped[dt.datetime] = _now()
@@ -322,6 +325,10 @@ class Outbox(Base):
state: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("'undo_window'"))
send_after: Mapped[dt.datetime] = mapped_column(TIMESTAMP(timezone=True), nullable=False)
error: Mapped[str | None] = mapped_column(Text)
# Retries after a send the server could not take (0021): state "retrying".
attempts: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0"))
next_attempt_at: Mapped[dt.datetime | None] = mapped_column(TIMESTAMP(timezone=True))
retry_until: Mapped[dt.datetime | None] = mapped_column(TIMESTAMP(timezone=True))
class SavedSearch(Base):
+89 -25
View File
@@ -8,14 +8,13 @@ to Sent (skipped for Gmail, which self-appends), harvest recipients into
from __future__ import annotations
import contextlib
import logging
import uuid
from datetime import UTC, datetime
from datetime import UTC, datetime, timedelta
from email.utils import getaddresses
from typing import Any
from sqlalchemy import select
from sqlalchemy import select, update
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession
@@ -27,7 +26,9 @@ from app.outbox.attachments import load_attachments
from app.outbox.mime_build import build_mime
from app.outbox.service import delete_mirrored_draft_row
from app.providers.base import RemoteMessage
from app.queue import enqueue_job
from app.search.factory import get_search_backend
from app.sync.health import is_temporary
from app.sync.provider_factory import build_provider
logger = logging.getLogger(__name__)
@@ -111,14 +112,80 @@ async def _persist_sent_copy(user_id: uuid.UUID, account: Account, raw: bytes, u
await bus.publish("thread.updated", {"user_id": str(user_id), "thread_id": str(thread_id)})
# A send the mail server could not take (unreachable, a temporary 4xx) is tried
# again after these gaps, then hourly, until RETRY_WINDOW after the first
# failure. Meanwhile the message waits in the Scheduled folder.
_RETRY_GAPS_S = (60, 120, 300, 600, 1800)
_RETRY_HOURLY_S = 3600
RETRY_WINDOW = timedelta(hours=24)
def _retry_gap(attempts: int) -> int:
return _RETRY_GAPS_S[attempts - 1] if attempts <= len(_RETRY_GAPS_S) else _RETRY_HOURLY_S
async def _send_failed(
oid: uuid.UUID, user_id: uuid.UUID, subject: str, exc: BaseException
) -> None:
"""Record a failed send: "retrying" if it may work later, else "failed"."""
now = datetime.now(UTC)
async with session_scope() as session:
ob = await session.get(Outbox, oid)
if ob is None:
return
ob.attempts += 1
ob.error = (getattr(exc, "detail", None) or str(exc) or type(exc).__name__)[:500]
ob.retry_until = ob.retry_until or now + RETRY_WINDOW
if is_temporary(exc) and now < ob.retry_until:
ob.state = "retrying"
ob.next_attempt_at = now + timedelta(seconds=_retry_gap(ob.attempts))
else:
ob.state = "failed"
ob.next_attempt_at = None
state, attempts, next_at, error = ob.state, ob.attempts, ob.next_attempt_at, ob.error
if state == "retrying" and next_at is not None:
await enqueue_job(
"send_job", str(oid), _defer_until=next_at, _job_id=f"send:{oid}:{attempts}"
)
await get_event_bus().publish(
"outbox.state_changed",
{
"user_id": str(user_id),
"outbox_id": str(oid),
"state": state,
"attempts": attempts,
"subject": subject,
"error": error,
"next_attempt_at": next_at.isoformat() if next_at else None,
},
)
if state == "failed":
logger.warning("send failed for outbox %s: %s", oid, error)
else:
logger.info(
"send of outbox %s failed (try %d), retrying at %s: %s", oid, attempts, next_at, error
)
async def send_job(ctx: dict[str, Any], outbox_id: str) -> None:
"""Send a queued message unless it was cancelled during the undo window."""
"""Send a queued message unless it was cancelled during the undo window.
Also runs each retry of a send that failed for a temporary reason. The row is
claimed with one conditional UPDATE, so two jobs for the same send (a planned
retry and "Retry now") can never both send it.
"""
oid = uuid.UUID(outbox_id)
async with session_scope() as session:
outbox = await session.get(Outbox, oid)
if outbox is None or outbox.state != "undo_window":
claimed = await session.scalar(
update(Outbox)
.where(Outbox.id == oid, Outbox.state.in_(("undo_window", "retrying")))
.values(state="sending")
.returning(Outbox.id)
)
if claimed is None:
return # cancelled, or already handled -> no SMTP traffic
outbox.state = "sending"
outbox = await session.get(Outbox, oid)
assert outbox is not None
draft = await session.get(Draft, outbox.draft_id)
if draft is None:
outbox.state = "failed"
@@ -153,28 +220,25 @@ async def send_job(ctx: dict[str, Any], outbox_id: str) -> None:
gmail = _is_gmail(account)
account_obj = account
subject = str(draft.payload.get("subject") or "")
provider = build_provider(account_obj)
appended_uid = 0
try:
await provider.send(transmit_raw, envelope_from, envelope_to)
if sent_path and not gmail:
appended_uid = await provider.append(sent_path, sent_raw, frozenset({"\\Seen"}))
# Remove the mirrored Drafts-folder copy now that it has been sent.
if drafts_path and draft_imap_uid:
with contextlib.suppress(Exception):
try:
await provider.send(transmit_raw, envelope_from, envelope_to)
except Exception as exc: # noqa: BLE001 - retry or record; the draft is kept
await _send_failed(oid, user_id, subject, exc)
return
# Delivered. Filing it is housekeeping: if the IMAP server is not there
# right now, the send still succeeded (the Sent poll files it later).
try:
if sent_path and not gmail:
appended_uid = await provider.append(sent_path, sent_raw, frozenset({"\\Seen"}))
# Remove the mirrored Drafts-folder copy now that it has been sent.
if drafts_path and draft_imap_uid:
await provider.delete_messages(drafts_path, [draft_imap_uid])
except Exception as exc: # noqa: BLE001 - record and notify, keep the draft
async with session_scope() as session:
ob = await session.get(Outbox, oid)
if ob is not None:
ob.state = "failed"
ob.error = str(exc)[:500]
await get_event_bus().publish(
"outbox.state_changed",
{"user_id": str(user_id), "outbox_id": str(oid), "state": "failed"},
)
logger.exception("send failed for outbox %s", oid)
return
except Exception as exc: # noqa: BLE001
logger.warning("outbox %s sent, but filing it on the server failed: %s", oid, exc)
finally:
await provider.close()
+17
View File
@@ -64,3 +64,20 @@ class SendResponse(BaseModel):
class OutboxStateOut(BaseModel):
outbox_id: uuid.UUID
state: str
class OutboxItemOut(BaseModel):
"""A send waiting in the Scheduled folder: being retried, or failed for good."""
outbox_id: uuid.UUID
draft_id: uuid.UUID
account_id: uuid.UUID
state: str # "retrying" | "failed"
attempts: int
error: str | None
next_attempt_at: str | None
retry_until: str | None
sent_at: str # when Send was pressed (the end of the undo window)
to: list[str]
subject: str
preview: str
+11 -1
View File
@@ -5,7 +5,7 @@ from __future__ import annotations
import uuid
from datetime import UTC, datetime, timedelta
from sqlalchemy import select
from sqlalchemy import ColumnElement, exists, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.config import get_settings
@@ -78,6 +78,16 @@ async def upsert_draft(
return draft
# A draft with one of these sends is in flight — or waiting in Scheduled for a
# retry — so it is not listed (or counted) in Drafts.
OUTBOX_ACTIVE = ("undo_window", "sending", "retrying", "failed")
def not_being_sent() -> ColumnElement[bool]:
"""SQL condition: this Draft row has no send in flight or waiting."""
return ~exists().where(Outbox.draft_id == Draft.id, Outbox.state.in_(OUTBOX_ACTIVE))
async def create_outbox(session: AsyncSession, *, user_id: uuid.UUID, draft: Draft) -> Outbox:
"""Create an outbox row in the undo window for a draft."""
send_after = datetime.now(UTC) + timedelta(seconds=get_settings().undo_seconds)
+5
View File
@@ -29,6 +29,11 @@ class ProviderConnectError(ProviderError):
"""Could not connect / TLS / greeting failure."""
class ProviderRejectedError(ProviderError):
"""The server refused for good (a permanent 5xx: recipient unknown, message
rejected). Trying again would only be refused again."""
class ProviderNotSupported(ProviderError):
"""A capability (e.g. IDLE) is not supported by the server."""
+65 -10
View File
@@ -26,7 +26,9 @@ from app.providers.base import (
MailProvider,
ProviderAuthError,
ProviderConnectError,
ProviderError,
ProviderNotSupported,
ProviderRejectedError,
RemoteMailbox,
RemoteMessage,
)
@@ -43,7 +45,10 @@ _FETCH_ITEMS = "(UID FLAGS INTERNALDATE RFC822.SIZE BODY.PEEK[])"
# Gmail (X-GM-EXT-1): also pull each message's labels so custom labels can map to
# chips without visiting every label folder (SPEC: Gmail sync / no-duplication).
_FETCH_ITEMS_GM = "(UID FLAGS INTERNALDATE RFC822.SIZE X-GM-LABELS BODY.PEEK[])"
_IDLE_TIMEOUT = 25 * 60 # SPEC §5.1: re-issue IDLE at least every 25 min
# SPEC §5.1 asks for IDLE to be re-issued at least every 25 min; 10 also means a
# connection that died silently (no FIN, e.g. the network dropped) is noticed
# within ~11 min — the re-issue finds it — instead of ~29.
_IDLE_TIMEOUT = 10 * 60
_CONNECT_TIMEOUT = 20
_PER_ACCOUNT_CONN_CAP = 4 # SPEC §6.3 / §4: IDLE + transient + short-lived jobs
_APPENDUID_RE = re.compile(rb"APPENDUID\s+\d+\s+(\d+)")
@@ -51,6 +56,52 @@ _COPYUID_RE = re.compile(rb"COPYUID\s+\d+\s+\S+\s+(\d+)")
_IMAP_MONTHS = ("Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec")
class _KeepConnectTask:
"""aioimaplib starts the TCP connection as a task it never awaits: a refused
or unreachable server then surfaces only as a 20 s greeting timeout, plus a
"Task exception was never retrieved" traceback in the log. Keep the task, so
``_connect`` can await it — failing at once, with the real reason."""
connect_task: asyncio.Task[object]
def create_client( # type: ignore[no-untyped-def]
self, host, port, loop, conn_lost_cb=None, ssl_context=None
) -> None:
local_loop = loop if loop is not None else asyncio.get_running_loop()
self.protocol = aioimaplib.IMAP4ClientProtocol(local_loop, conn_lost_cb)
self.connect_task = local_loop.create_task(
local_loop.create_connection(lambda: self.protocol, host, port, ssl=ssl_context)
)
class _IMAP4(_KeepConnectTask, aioimaplib.IMAP4): # type: ignore[misc]
pass
class _IMAP4_SSL(_KeepConnectTask, aioimaplib.IMAP4_SSL): # type: ignore[misc] # noqa: N801
pass
def _smtp_error(exc: aiosmtplib.SMTPException) -> ProviderError:
"""An SMTP failure as the provider error that says whether to try again.
A 5xx reply is the server's final word (recipient unknown, message refused):
``ProviderRejectedError``, or ``ProviderAuthError`` for a refused login. Any
4xx reply, a timeout or a dropped connection is temporary:
``ProviderConnectError``, and the send is retried later.
"""
if isinstance(exc, aiosmtplib.SMTPRecipientsRefused):
codes = [r.code for r in exc.recipients]
if codes and all(c >= 500 for c in codes):
return ProviderRejectedError(f"recipients refused: {exc}")
return ProviderConnectError(f"SMTP send failed: {exc}")
if isinstance(exc, aiosmtplib.SMTPResponseException) and exc.code >= 500:
if isinstance(exc, aiosmtplib.SMTPAuthenticationError):
return ProviderAuthError(f"SMTP auth failed: {exc}")
return ProviderRejectedError(f"SMTP refused: {exc}")
return ProviderConnectError(f"SMTP send failed: {exc}")
def _imap_date(d: date) -> str:
"""RFC 3501 search date, e.g. ``5-Jul-2026`` (locale-independent)."""
return f"{d.day}-{_IMAP_MONTHS[d.month - 1]}-{d.year}"
@@ -133,19 +184,22 @@ class ImapSmtpProvider(MailProvider):
await self._acquire_conn_slots()
try:
try:
imap: aioimaplib.IMAP4
if cfg.imap_tls == "ssl":
imap = aioimaplib.IMAP4_SSL(
imap = _IMAP4_SSL(
host=cfg.imap_host, port=cfg.imap_port, timeout=_CONNECT_TIMEOUT
)
else:
imap = aioimaplib.IMAP4(
host=cfg.imap_host, port=cfg.imap_port, timeout=_CONNECT_TIMEOUT
)
imap = _IMAP4(host=cfg.imap_host, port=cfg.imap_port, timeout=_CONNECT_TIMEOUT)
# Refused / unknown host: fails here at once. No answer at all:
# gives up after the timeout (the connect attempt is cancelled).
await asyncio.wait_for(imap.connect_task, _CONNECT_TIMEOUT)
await imap.wait_hello_from_server()
if cfg.imap_tls == "starttls":
await imap.starttls()
except (OSError, TimeoutError) as exc:
raise ProviderConnectError(f"cannot connect to {cfg.imap_host}: {exc}") from exc
why = str(exc) or f"no answer within {_CONNECT_TIMEOUT} s"
raise ProviderConnectError(f"cannot connect to {cfg.imap_host}: {why}") from exc
resp = await imap.login(cfg.username, cfg.password)
if resp.result != "OK":
@@ -391,7 +445,10 @@ class ImapSmtpProvider(MailProvider):
await self._select(mailbox)
idle_task = await imap.idle_start(timeout=_IDLE_TIMEOUT)
try:
await imap.wait_server_push()
# No word from the server even after IDLE's own end: just return, and
# the caller's poll shows whether the connection is still there.
with contextlib.suppress(TimeoutError):
await imap.wait_server_push(timeout=_IDLE_TIMEOUT + 60)
finally:
imap.idle_done()
with contextlib.suppress(Exception):
@@ -473,7 +530,5 @@ class ImapSmtpProvider(MailProvider):
start_tls=start_tls,
timeout=_CONNECT_TIMEOUT,
)
except aiosmtplib.SMTPAuthenticationError as exc:
raise ProviderAuthError(f"SMTP auth failed: {exc}") from exc
except aiosmtplib.SMTPException as exc:
raise ProviderConnectError(f"SMTP send failed: {exc}") from exc
raise _smtp_error(exc) from exc
+106 -50
View File
@@ -1,9 +1,11 @@
"""Outbound user actions: flag changes and moves (SPEC §6.3).
The API writes the change to the DB first, then enqueues a job here to push it to
the mail server. On provider failure the job retries (arq, ×5 exponential) and, on
final failure, reverts the DB change and emits ``flags.changed`` with
``reverted=true`` so the UI can roll back.
the mail server. While the server cannot be reached the job is retried (arq
``Retry``) with growing gaps for about 24 hours (``ACTION_MAX_TRIES``); a flag
change that still cannot be pushed is then reverted in the DB and announced with
``flags.changed`` / ``reverted=true`` so the UI rolls back. A plain exception does
NOT make arq retry — only ``Retry`` does.
"""
from __future__ import annotations
@@ -15,6 +17,7 @@ from collections.abc import Iterable
from datetime import UTC, datetime, timedelta
from typing import Any
from arq import Retry
from sqlalchemy import delete, func, select, update
from sqlalchemy.ext.asyncio import AsyncSession
@@ -22,6 +25,7 @@ from app.config import get_settings
from app.db import session_scope
from app.events import get_event_bus
from app.models import Account, Draft, Mailbox, Message, Thread
from app.sync.health import is_temporary
from app.sync.provider_factory import build_provider
from app.threading_.threader import denormalize_thread
@@ -30,7 +34,26 @@ logger = logging.getLogger(__name__)
# User-settable IMAP flags (SPEC §8.2 is:read/starred/answered). \Deleted/\Draft
# are managed by move/expunge and drafts, not this surface.
ALLOWED_FLAGS = frozenset({"\\Seen", "\\Flagged", "\\Answered"})
_MAX_TRIES = 5
# Gaps between tries while the server is unreachable: ~24 hours in all.
_RETRY_DEFERS_S = (30, 60, 120, 300, 600, 1800, 3600, 7200, 14400, 28800, 28800)
ACTION_MAX_TRIES = len(_RETRY_DEFERS_S) + 1
def _retry_later(ctx: dict[str, Any], exc: BaseException) -> Retry | None:
"""A ``Retry`` for a temporary failure while tries remain, else None (give up)."""
if not is_temporary(exc):
return None
attempt = int(ctx.get("job_try", 1))
if attempt >= ACTION_MAX_TRIES:
return None
logger.info(
"server unreachable (try %d), retrying in %ds: %s",
attempt,
_RETRY_DEFERS_S[attempt - 1],
exc,
)
return Retry(defer=_RETRY_DEFERS_S[attempt - 1])
# Folder names to create when a target special-use folder is missing on the server.
_DEFAULT_FOLDER_NAME = {"archive": "Archive", "trash": "Trash", "junk": "Spam"}
@@ -71,6 +94,8 @@ async def push_flags_job(
message = await session.get(Message, mid)
if message is None:
return
if _superseded(message.flags, add, remove):
return # changed again since: that later change's own job pushes it
account = await session.get(Account, message.account_id)
mailbox = await session.get(Mailbox, message.mailbox_id)
if account is None or mailbox is None:
@@ -80,21 +105,29 @@ async def push_flags_job(
provider = build_provider(account)
try:
await provider.push_flags(path, uid, frozenset(add), frozenset(remove))
except Exception:
if ctx.get("job_try", 1) >= _MAX_TRIES:
await _revert_flags(mid, user_id, add, remove)
return
raise # arq retries with backoff
except Exception as exc:
retry = _retry_later(ctx, exc)
if retry is not None:
raise retry from exc
logger.warning("flag change for message %s could not be pushed: %s", mid, exc)
await _revert_flags(mid, user_id, add, remove)
finally:
await provider.close()
def _superseded(flags: list[str], add: list[str], remove: list[str]) -> bool:
"""The message's flags no longer show this change: a later one replaced it
(e.g. read, then unread again while the server was unreachable)."""
now = set(flags)
return not set(add) <= now or bool(set(remove) & now)
async def _revert_flags(
message_id: uuid.UUID, user_id: uuid.UUID, add: list[str], remove: list[str]
) -> None:
async with session_scope() as session:
message = await session.get(Message, message_id)
if message is None:
if message is None or _superseded(message.flags, add, remove):
return
# Undo the change: remove what we added, add back what we removed.
message.flags = sorted((set(message.flags) - set(add)) | set(remove))
@@ -138,11 +171,12 @@ async def move_job(
provider = build_provider(account)
try:
new_uid = await provider.move(uid, src_path, dst_path)
except Exception:
if ctx.get("job_try", 1) >= _MAX_TRIES:
logger.warning("move failed for message %s; leaving optimistic state", mid)
return
raise
except Exception as exc:
retry = _retry_later(ctx, exc)
if retry is not None:
raise retry from exc
logger.warning("move failed for message %s; leaving optimistic state: %s", mid, exc)
return
finally:
await provider.close()
@@ -193,28 +227,39 @@ async def move_thread_to_folder_job(ctx: dict[str, Any], thread_id: str, special
# (message_id, account_id, target_path, new_uid) applied after the network moves.
moves: list[tuple[uuid.UUID, uuid.UUID, str, int]] = []
to_create: list[tuple[uuid.UUID, str, int]] = [] # account_id, path, uidvalidity
unreachable: BaseException | None = None # set: retry the moves that did not happen
for account_id in account_ids:
provider = build_provider(accounts[account_id])
try:
if account_id in target_id_path:
target_id, target_path = target_id_path[account_id]
else:
target_id, target_path = None, _DEFAULT_FOLDER_NAME.get(special_use, special_use)
await provider.create_mailbox(target_path)
remote = await provider.list_mailboxes()
uidv = next((r.uidvalidity for r in remote if r.path == target_path), 0)
to_create.append((account_id, target_path, uidv))
for mid, m_account, m_mailbox, m_uid in snapshot:
if m_account != account_id or m_mailbox == target_id:
continue
src_path = path_by_id.get(m_mailbox, "INBOX")
try:
new_uid = await provider.move(m_uid, src_path, target_path)
except Exception:
logger.exception("move failed for message %s -> %s", mid, target_path)
continue
moves.append((mid, account_id, target_path, new_uid))
try:
if account_id in target_id_path:
target_id, target_path = target_id_path[account_id]
else:
target_id, target_path = (
None,
_DEFAULT_FOLDER_NAME.get(special_use, special_use),
)
await provider.create_mailbox(target_path)
remote = await provider.list_mailboxes()
uidv = next((r.uidvalidity for r in remote if r.path == target_path), 0)
to_create.append((account_id, target_path, uidv))
for mid, m_account, m_mailbox, m_uid in snapshot:
if m_account != account_id or m_mailbox == target_id:
continue
src_path = path_by_id.get(m_mailbox, "INBOX")
try:
new_uid = await provider.move(m_uid, src_path, target_path)
except Exception as exc:
if is_temporary(exc):
raise # the server is gone: stop this account, retry it later
logger.exception("move failed for message %s -> %s", mid, target_path)
continue
moves.append((mid, account_id, target_path, new_uid))
except Exception as exc:
if not is_temporary(exc):
raise
unreachable = exc
finally:
await provider.close()
@@ -246,6 +291,10 @@ async def move_thread_to_folder_job(ctx: dict[str, Any], thread_id: str, special
await get_event_bus().publish(
"thread.updated", {"user_id": str(user_id), "thread_id": str(tid)}
)
# What moved is recorded; the rest goes again once the server is back (the
# re-run skips messages already in the target folder).
if unreachable is not None and (retry := _retry_later(ctx, unreachable)) is not None:
raise retry from unreachable
async def restore_thread_job(ctx: dict[str, Any], thread_id: str) -> None:
@@ -270,9 +319,7 @@ async def restore_thread_job(ctx: dict[str, Any], thread_id: str) -> None:
await session.scalars(select(Mailbox).where(Mailbox.account_id.in_(account_ids)))
)
path_by_id = {mb.id: mb.path for mb in mailboxes}
inbox_by_account = {
mb.account_id: mb.id for mb in mailboxes if mb.special_use == "inbox"
}
inbox_by_account = {mb.account_id: mb.id for mb in mailboxes if mb.special_use == "inbox"}
# (message_id, account_id, source_path, target_id) — resolve the target now.
plan: list[tuple[uuid.UUID, uuid.UUID, str, uuid.UUID, int]] = []
for m in messages:
@@ -287,6 +334,7 @@ async def restore_thread_job(ctx: dict[str, Any], thread_id: str) -> None:
# (message_id, target_id, new_uid) applied after the network moves.
moves: list[tuple[uuid.UUID, uuid.UUID, int]] = []
unreachable: BaseException | None = None
for account_id in account_ids:
provider = build_provider(accounts[account_id])
try:
@@ -296,7 +344,10 @@ async def restore_thread_job(ctx: dict[str, Any], thread_id: str) -> None:
target_path = path_by_id[target_id]
try:
new_uid = await provider.move(uid, src_path, target_path)
except Exception:
except Exception as exc:
if is_temporary(exc):
unreachable = exc # the server is gone: stop this account
break
logger.exception("restore move failed for message %s -> %s", mid, target_path)
continue
moves.append((mid, target_id, new_uid))
@@ -314,6 +365,8 @@ async def restore_thread_job(ctx: dict[str, Any], thread_id: str) -> None:
await get_event_bus().publish(
"thread.updated", {"user_id": str(user_id), "thread_id": str(tid)}
)
if unreachable is not None and (retry := _retry_later(ctx, unreachable)) is not None:
raise retry from unreachable
async def delete_thread_job(ctx: dict[str, Any], thread_id: str) -> None:
@@ -352,8 +405,12 @@ async def delete_thread_job(ctx: dict[str, Any], thread_id: str) -> None:
continue
try:
await provider.delete_messages(path, uids)
except Exception:
logger.exception("expunge failed for %s in %s", uids, path)
except Exception as exc:
# Keep the rows while the server is away: deleted here but
# not there, the next sync would bring the messages back.
if (retry := _retry_later(ctx, exc)) is not None:
raise retry from exc
logger.warning("expunge failed for %s in %s: %s", uids, path, exc)
finally:
await provider.close()
@@ -389,9 +446,7 @@ async def empty_folder_job(ctx: dict[str, Any], user_id: str, special_use: str)
async with session_scope() as session:
mailboxes = list(
await session.scalars(
select(Mailbox).where(
Mailbox.user_id == uid, Mailbox.special_use == special_use
)
select(Mailbox).where(Mailbox.user_id == uid, Mailbox.special_use == special_use)
)
)
if not mailboxes:
@@ -423,8 +478,11 @@ async def empty_folder_job(ctx: dict[str, Any], user_id: str, special_use: str)
continue
try:
await provider.delete_messages(path, uids)
except Exception:
logger.exception("empty folder: expunge failed for %s in %s", uids, path)
except Exception as exc:
# As for a thread: keep the rows until the server has them gone.
if (retry := _retry_later(ctx, exc)) is not None:
raise retry from exc
logger.warning("empty folder: expunge failed for %s in %s: %s", uids, path, exc)
finally:
await provider.close()
@@ -520,9 +578,9 @@ async def autopurge_trash_spam(ctx: dict[str, Any]) -> None:
user_ids = {r.user_id for r in stale}
async with session_scope() as session:
await session.execute(
update(Draft).where(Draft.reply_to_message_id.in_(msg_ids)).values(
reply_to_message_id=None
)
update(Draft)
.where(Draft.reply_to_message_id.in_(msg_ids))
.values(reply_to_message_id=None)
)
await session.execute(delete(Message).where(Message.id.in_(msg_ids)))
await session.flush()
@@ -539,6 +597,4 @@ async def autopurge_trash_spam(ctx: dict[str, Any]) -> None:
bus = get_event_bus()
for user_id in user_ids:
await bus.publish("message.deleted", {"user_id": str(user_id)})
logger.info(
"autopurge: removed %d Trash/Spam message(s) older than %dd", len(msg_ids), days
)
logger.info("autopurge: removed %d Trash/Spam message(s) older than %dd", len(msg_ids), days)
+39 -10
View File
@@ -18,10 +18,11 @@ from app.config import get_settings
from app.events import get_event_bus
from app.indexer.indexer import PersistResult, persist_message
from app.models import Account, Draft, Mailbox, Message
from app.providers.base import MailboxDelta, ProviderNotSupported
from app.providers.base import MailboxDelta, ProviderAuthError, ProviderNotSupported
from app.providers.imap_smtp import ImapSmtpProvider
from app.search.factory import get_search_backend
from app.sync.folder_names import infer_special_use
from app.sync.health import AccountHealth, is_temporary
from app.sync.provider_factory import build_provider
from app.threading_.threader import denormalize_thread
@@ -58,6 +59,12 @@ async def _set_state(account_id: uuid.UUID, state: str, detail: str = "") -> Non
await session.execute(
update(Account).where(Account.id == account_id).values(state=state, state_detail=detail)
)
if state == "live": # (re)connected and synced: no sync problem any more
await session.execute(
update(Account)
.where(Account.id == account_id)
.values(sync_error=None, sync_error_since=None)
)
# Keep the sync:live registry in step with the account's state so the sync
# coordinator claims it once live and drops it otherwise (refactor §2).
from app.redis_client import get_redis
@@ -504,7 +511,10 @@ async def _reconcile_drafts(
async def live_loop(
account_id: uuid.UUID, mailbox_path: str = "INBOX", stop: asyncio.Event | None = None
account_id: uuid.UUID,
mailbox_path: str = "INBOX",
stop: asyncio.Event | None = None,
health: AccountHealth | None = None,
) -> None:
"""Maintain a live IDLE/poll loop for one mailbox (SPEC §6.3)."""
from app.db import session_scope
@@ -518,6 +528,9 @@ async def live_loop(
provider = build_provider(account_obj)
try:
await provider.verify() # signed in: the account is reachable
if health is not None:
await health.ok()
while not stop.is_set():
try:
await provider.idle(mailbox_path)
@@ -544,7 +557,9 @@ async def live_loop(
await provider.close()
async def poll_folders_once(account_id: uuid.UUID, *, include_inbox: bool) -> None:
async def poll_folders_once(
account_id: uuid.UUID, *, include_inbox: bool, health: AccountHealth | None = None
) -> None:
"""One CONDSTORE poll pass over an account's special-use folders.
``include_inbox`` also polls INBOX (used by the manual Refresh); the periodic
@@ -586,14 +601,22 @@ async def poll_folders_once(account_id: uuid.UUID, *, include_inbox: bool) -> No
highestmodseq=mailbox_obj.highestmodseq,
)
await apply_delta(provider, account_obj, mailbox_obj, delta)
except Exception:
except Exception as exc:
# Server gone, or the sign-in refused: the other folders would only
# fail the same way (20 s each). End the pass; the caller decides.
if is_temporary(exc) or isinstance(exc, ProviderAuthError):
raise
logger.exception("poll failed: account=%s mailbox=%s", account_id, mailbox_obj.path)
if health is not None:
await health.ok()
finally:
await provider.close()
async def _poll_secondary_once(account_id: uuid.UUID) -> None:
await poll_folders_once(account_id, include_inbox=False)
async def _poll_secondary_once(
account_id: uuid.UUID, health: AccountHealth | None = None
) -> None:
await poll_folders_once(account_id, include_inbox=False, health=health)
async def refresh_account_job(ctx: dict[str, Any], account_id: str) -> None:
@@ -602,7 +625,10 @@ async def refresh_account_job(ctx: dict[str, Any], account_id: str) -> None:
async def secondary_poll_loop(
account_id: uuid.UUID, stop: asyncio.Event | None = None, interval: float | None = None
account_id: uuid.UUID,
stop: asyncio.Event | None = None,
interval: float | None = None,
health: AccountHealth | None = None,
) -> None:
"""Periodically CONDSTORE-poll non-INBOX folders (Sent/Drafts/Archive/Trash/Junk).
@@ -624,11 +650,14 @@ async def secondary_poll_loop(
await asyncio.wait_for(stop.wait(), timeout=wait)
if stop.is_set():
break
await _poll_secondary_once(account_id)
await _poll_secondary_once(account_id, health)
async def poll_all_loop(
account_id: uuid.UUID, interval: float, stop: asyncio.Event | None = None
account_id: uuid.UUID,
interval: float,
stop: asyncio.Event | None = None,
health: AccountHealth | None = None,
) -> None:
"""Transient CONDSTORE poll of INBOX + special folders every ``interval`` seconds.
@@ -646,4 +675,4 @@ async def poll_all_loop(
await asyncio.wait_for(stop.wait(), timeout=wait)
if stop.is_set():
break
await poll_folders_once(account_id, include_inbox=True)
await poll_folders_once(account_id, include_inbox=True, health=health)
+109
View File
@@ -0,0 +1,109 @@
"""An account's connection health.
Telling "the mail server is not there right now" from everything else.
``is_temporary`` decides whether a failure is worth another try later: the
server could not be reached, timed out, dropped the connection, or answered
"not now" (SMTP 4xx). Sends, user actions and the sync loops all retry on it,
and only on it — a refused password or a permanent 5xx is not retried.
"""
from __future__ import annotations
import logging
import uuid
from datetime import UTC, datetime
from aioimaplib import aioimaplib
from sqlalchemy import update
from app.providers.base import ProviderAuthError, ProviderConnectError
logger = logging.getLogger(__name__)
def is_temporary(exc: BaseException) -> bool:
if isinstance(exc, (ProviderConnectError, TimeoutError, OSError)):
return True
# aioimaplib's own: a command that timed out, or a reply cut off.
if isinstance(exc, (aioimaplib.CommandTimeout, aioimaplib.IncompleteRead)):
return True
# After the connection drops, aioimaplib refuses every further command as
# "illegal in state LOGOUT" — the connection is gone, not the command wrong.
return isinstance(exc, aioimaplib.Abort) and "illegal in state" in str(exc)
# Failed connections in a row before an account shows as unreachable, and
# refused sign-ins in a row before its syncing stops ("auth_failed": retrying a
# wrong password only gets the account locked by the provider). Changing the
# password in Settings re-runs the initial sync, which sets it live again.
UNREACHABLE_AFTER = 3
AUTH_GIVE_UP_AFTER = 3
class AccountHealth:
"""One account's run of failures, shared by its sync loops."""
def __init__(self, account_id: uuid.UUID) -> None:
self.account_id = account_id
self.failures = 0
self.auth_failures = 0
# Whether the account is marked unreachable; None = not known yet (a
# mark can survive a worker restart), so the first success checks.
self.marked: bool | None = None
# Set once the sign-in was refused too often: the account is no longer
# synced (state auth_failed), so its loops should stop now.
self.gave_up = False
async def ok(self) -> None:
"""The server answered: reset the counts, and clear an unreachable mark."""
self.failures = 0
self.auth_failures = 0
if self.marked is not False:
await _set_sync_error(self.account_id, None)
self.marked = False
async def failed(self, exc: BaseException) -> None:
if isinstance(exc, ProviderAuthError):
self.auth_failures += 1
if self.auth_failures >= AUTH_GIVE_UP_AFTER:
from app.sync.engine import _set_state
logger.warning(
"account %s: sign-in refused %d times; stopping sync",
self.account_id,
self.auth_failures,
)
await _set_state(self.account_id, "auth_failed", exc.detail[:500])
self.gave_up = True
return
if not is_temporary(exc):
return
self.failures += 1
if self.failures >= UNREACHABLE_AFTER and self.marked is not True:
await _set_sync_error(self.account_id, "unreachable")
self.marked = True
async def _set_sync_error(account_id: uuid.UUID, error: str | None) -> None:
"""Mark (or clear) an account's sync problem; tell the user's open apps."""
from app.db import session_scope
from app.events import get_event_bus
from app.models import Account
now = datetime.now(UTC)
async with session_scope() as session:
cond = Account.sync_error.isnot(None) if error is None else Account.sync_error.is_(None)
user_id = await session.scalar(
update(Account)
.where(Account.id == account_id, cond)
.values(sync_error=error, sync_error_since=None if error is None else now)
.returning(Account.user_id)
)
if user_id is None:
return # nothing changed
logger.warning("account %s: %s", account_id, error or "reachable again")
await get_event_bus().publish(
"account.state_changed",
{"user_id": str(user_id), "account_id": str(account_id), "sync_error": error},
)
+53 -7
View File
@@ -18,6 +18,7 @@ from __future__ import annotations
import asyncio
import contextlib
import logging
import random
import uuid
from collections.abc import Awaitable, Callable
@@ -26,11 +27,18 @@ from sqlalchemy import select
from app.config import get_settings
from app.db import session_scope
from app.models import Account
from app.providers.base import ProviderAuthError
from app.sync.engine import live_loop, poll_all_loop, secondary_poll_loop
from app.sync.health import AccountHealth, is_temporary
logger = logging.getLogger(__name__)
# After a loop fails: wait 30 s, doubling with each failure in a row up to 10 min
# (±20%, so accounts on one dead server don't retry in step). A loop that ran a
# good while before failing starts again from 30 s.
_RECONNECT_BACKOFF = 30.0
_MAX_BACKOFF = 600.0
_HEALTHY_RUN_S = 300.0
_VERIFY_INTERVAL = 60.0
Tier = str # "hot" | "warm" | "cold"
@@ -49,6 +57,8 @@ class AccountRunner:
# coordinator reaps such runners and releases their lease.
self.dead = asyncio.Event()
self.verify_interval = _VERIFY_INTERVAL
# Failures in a row across this account's loops: unreachable / sign-in failed.
self.health = AccountHealth(account_id)
def start(self) -> None:
"""Launch the verify task and the current tier's work loops."""
@@ -98,13 +108,15 @@ class AccountRunner:
await asyncio.gather(*tasks, return_exceptions=True)
async def _run_live(self) -> None:
await live_loop(self.account_id, "INBOX", self._stop)
await live_loop(self.account_id, "INBOX", self._stop, self.health)
async def _run_secondary(self) -> None:
await secondary_poll_loop(self.account_id, self._stop, get_settings().poll_warm)
await secondary_poll_loop(
self.account_id, self._stop, get_settings().poll_warm, self.health
)
async def _run_poll(self, interval: float) -> None:
await poll_all_loop(self.account_id, interval, self._stop)
await poll_all_loop(self.account_id, interval, self._stop, self.health)
async def _verify_loop(self) -> None:
"""Periodically confirm the account is still live; flag ``dead`` if not.
@@ -129,13 +141,47 @@ class AccountRunner:
return state == "live"
async def _supervise(self, run: Callable[[], Awaitable[None]], label: str) -> None:
"""Run a loop, restarting it (with backoff) until stopped or cancelled."""
"""Run a loop, restarting it (with growing backoff) until stopped or cancelled."""
loop = asyncio.get_running_loop()
fails = 0
while not self._stop.is_set():
started = loop.time()
try:
await run()
except asyncio.CancelledError:
raise
except Exception:
logger.exception("%s loop for account %s crashed; retrying", label, self.account_id)
except Exception as exc:
if loop.time() - started > _HEALTHY_RUN_S:
fails = 0 # it had been working: a fresh failure, not the same outage
fails += 1
delay = min(_MAX_BACKOFF, _RECONNECT_BACKOFF * 2 ** (fails - 1))
delay *= random.uniform(0.8, 1.2)
try:
await self.health.failed(exc)
except Exception: # noqa: BLE001 - never let bookkeeping kill the loop
logger.exception("account %s: recording the failure failed", self.account_id)
if self.health.gave_up:
# Sign-in refused too often: not synced any more (the health
# tracker logged why). Flag the runner so the coordinator
# reaps it and releases the lease.
self.dead.set()
return
if is_temporary(exc) or isinstance(exc, ProviderAuthError):
# Expected while a server is down or a password is wrong: one
# line, not a traceback every half minute.
logger.warning(
"%s loop for account %s: %s; retrying in %.0fs",
label,
self.account_id,
getattr(exc, "detail", None) or exc or type(exc).__name__,
delay,
)
else:
logger.exception(
"%s loop for account %s crashed; retrying in %.0fs",
label,
self.account_id,
delay,
)
with contextlib.suppress(TimeoutError):
await asyncio.wait_for(self._stop.wait(), timeout=_RECONNECT_BACKOFF)
await asyncio.wait_for(self._stop.wait(), timeout=delay)