diff --git a/backend/app/api/accounts.py b/backend/app/api/accounts.py
index f308f61..1bea548 100644
--- a/backend/app/api/accounts.py
+++ b/backend/app/api/accounts.py
@@ -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
]
diff --git a/backend/app/api/compose.py b/backend/app/api/compose.py
index 598c318..d11b515 100644
--- a/backend/app/api/compose.py
+++ b/backend/app/api/compose.py
@@ -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)}
diff --git a/backend/app/api/mailboxes.py b/backend/app/api/mailboxes.py
index 378e0f1..0a4aa81 100644
--- a/backend/app/api/mailboxes.py
+++ b/backend/app/api/mailboxes.py
@@ -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(
diff --git a/backend/app/api/schemas.py b/backend/app/api/schemas.py
index 7296f27..00058a8 100644
--- a/backend/app/api/schemas.py
+++ b/backend/app/api/schemas.py
@@ -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):
diff --git a/backend/app/models.py b/backend/app/models.py
index 8114973..522c540 100644
--- a/backend/app/models.py
+++ b/backend/app/models.py
@@ -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):
diff --git a/backend/app/outbox/jobs.py b/backend/app/outbox/jobs.py
index 87a492f..407e904 100644
--- a/backend/app/outbox/jobs.py
+++ b/backend/app/outbox/jobs.py
@@ -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()
diff --git a/backend/app/outbox/schemas.py b/backend/app/outbox/schemas.py
index 90f14d4..e1b2832 100644
--- a/backend/app/outbox/schemas.py
+++ b/backend/app/outbox/schemas.py
@@ -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
diff --git a/backend/app/outbox/service.py b/backend/app/outbox/service.py
index 6f2ea5d..9168dcb 100644
--- a/backend/app/outbox/service.py
+++ b/backend/app/outbox/service.py
@@ -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)
diff --git a/backend/app/providers/base.py b/backend/app/providers/base.py
index 4a57bed..1521305 100644
--- a/backend/app/providers/base.py
+++ b/backend/app/providers/base.py
@@ -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."""
diff --git a/backend/app/providers/imap_smtp/provider.py b/backend/app/providers/imap_smtp/provider.py
index 71e3409..be37bd8 100644
--- a/backend/app/providers/imap_smtp/provider.py
+++ b/backend/app/providers/imap_smtp/provider.py
@@ -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
diff --git a/backend/app/sync/actions.py b/backend/app/sync/actions.py
index 38ba0ad..a729cb9 100644
--- a/backend/app/sync/actions.py
+++ b/backend/app/sync/actions.py
@@ -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)
diff --git a/backend/app/sync/engine.py b/backend/app/sync/engine.py
index cb839f4..0ea8662 100644
--- a/backend/app/sync/engine.py
+++ b/backend/app/sync/engine.py
@@ -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)
diff --git a/backend/app/sync/health.py b/backend/app/sync/health.py
new file mode 100644
index 0000000..78e27fb
--- /dev/null
+++ b/backend/app/sync/health.py
@@ -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},
+ )
diff --git a/backend/app/sync/runner.py b/backend/app/sync/runner.py
index 17c4bef..25e448d 100644
--- a/backend/app/sync/runner.py
+++ b/backend/app/sync/runner.py
@@ -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)
diff --git a/backend/migrations/versions/0021_retries_and_account_health.py b/backend/migrations/versions/0021_retries_and_account_health.py
new file mode 100644
index 0000000..c683ed3
--- /dev/null
+++ b/backend/migrations/versions/0021_retries_and_account_health.py
@@ -0,0 +1,43 @@
+"""send retries for 24 hours, and an account's sync health
+
+A send the mail server could not take (SMTP unreachable, a temporary 4xx) is no
+longer failed at once: the outbox row goes to ``retrying`` and is tried again
+with growing gaps until ``retry_until`` (24 hours after the first failure).
+``attempts`` counts the tries, ``next_attempt_at`` is when the next one runs.
+
+An account whose server cannot be reached for a while carries ``sync_error``
+(``unreachable``) and ``sync_error_since``; both clear on the next success.
+
+Revision ID: 0021
+Revises: 0020
+Create Date: 2026-10-06
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from alembic import op
+
+revision: str = "0021"
+down_revision: str | None = "0020"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+_COLUMNS = (
+ ("outbox", "attempts", "integer NOT NULL DEFAULT 0"),
+ ("outbox", "next_attempt_at", "timestamptz"),
+ ("outbox", "retry_until", "timestamptz"),
+ ("accounts", "sync_error", "text"),
+ ("accounts", "sync_error_since", "timestamptz"),
+)
+
+
+def upgrade() -> None:
+ for table, name, ddl in _COLUMNS:
+ op.execute(f"ALTER TABLE {table} ADD COLUMN {name} {ddl}")
+
+
+def downgrade() -> None:
+ for table, name, _ in reversed(_COLUMNS):
+ op.execute(f"ALTER TABLE {table} DROP COLUMN IF EXISTS {name}")
diff --git a/backend/tests/mail/test_actions.py b/backend/tests/mail/test_actions.py
index 11a5d66..ce677f8 100644
--- a/backend/tests/mail/test_actions.py
+++ b/backend/tests/mail/test_actions.py
@@ -9,7 +9,9 @@ from datetime import UTC, datetime
from email.message import EmailMessage
import httpx
+import pytest
import pytest_asyncio
+from arq import Retry
from httpx import ASGITransport
from sqlalchemy import delete, func, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
@@ -23,7 +25,7 @@ from app.identity.sessions import create_session
from app.indexer.indexer import persist_message
from app.main import create_app
from app.models import Account, Mailbox, Message, Thread, User
-from app.providers.base import RemoteMessage
+from app.providers.base import ProviderConnectError, RemoteMessage
from app.sync import actions as sync_actions
CSRF = "csrf"
@@ -319,3 +321,103 @@ async def test_push_flags_job_reverts_on_failure(ctx: Ctx, monkeypatch) -> None:
reverted = await ctx.session.get(Message, mid)
await ctx.session.refresh(reverted)
assert reverted is not None and "\\Seen" not in reverted.flags
+
+
+# --- the mail server unreachable: real retries (arq Retry), ~24 hours -------------
+
+
+class _Unreachable:
+ """A provider whose server cannot be reached."""
+
+ def __init__(self) -> None:
+ self.calls = 0
+
+ async def push_flags(self, *a: object, **k: object) -> None:
+ self.calls += 1
+ raise ProviderConnectError("cannot connect to imap.example.com")
+
+ async def move(self, *a: object, **k: object) -> int:
+ self.calls += 1
+ raise ProviderConnectError("cannot connect to imap.example.com")
+
+ async def delete_messages(self, *a: object, **k: object) -> None:
+ self.calls += 1
+ raise ProviderConnectError("cannot connect to imap.example.com")
+
+ async def close(self) -> None:
+ return None
+
+
+async def _mark_read(ctx: Ctx) -> uuid.UUID:
+ mid = ctx.message_ids[0]
+ msg = await ctx.session.get(Message, mid)
+ assert msg is not None
+ msg.flags = ["\\Seen"]
+ await ctx.session.commit()
+ return mid
+
+
+async def test_flag_push_retries_while_unreachable(ctx: Ctx, monkeypatch) -> None: # type: ignore[no-untyped-def]
+ mid = await _mark_read(ctx)
+ monkeypatch.setattr(sync_actions, "build_provider", lambda _a: _Unreachable())
+
+ with pytest.raises(Retry) as caught:
+ await sync_actions.push_flags_job({"job_try": 1}, str(mid), ["\\Seen"], [])
+ assert caught.value.defer_score == 30_000 # first gap: 30 s
+ msg = await ctx.session.get(Message, mid)
+ await ctx.session.refresh(msg)
+ assert msg is not None and "\\Seen" in msg.flags # kept while retrying
+
+ # The last try still cannot reach it: the change is rolled back.
+ await sync_actions.push_flags_job(
+ {"job_try": sync_actions.ACTION_MAX_TRIES}, str(mid), ["\\Seen"], []
+ )
+ await ctx.session.refresh(msg)
+ assert "\\Seen" not in msg.flags
+
+
+def test_retry_gaps_cover_about_a_day() -> None:
+ total = sum(sync_actions._RETRY_DEFERS_S)
+ assert 23 * 3600 < total <= 24 * 3600
+
+
+async def test_superseded_flag_change_is_not_pushed(ctx: Ctx, monkeypatch) -> None: # type: ignore[no-untyped-def]
+ # "Read" was queued, then the message was marked unread again: the stale
+ # "read" must not reach the server after the newer change.
+ mid = ctx.message_ids[0]
+ provider = _Unreachable()
+ monkeypatch.setattr(sync_actions, "build_provider", lambda _a: provider)
+ await sync_actions.push_flags_job({"job_try": 3}, str(mid), ["\\Seen"], [])
+ assert provider.calls == 0
+
+
+async def test_archive_retries_while_unreachable(ctx: Ctx, monkeypatch) -> None: # type: ignore[no-untyped-def]
+ provider = _Unreachable()
+ monkeypatch.setattr(sync_actions, "build_provider", lambda _a: provider)
+ with pytest.raises(Retry):
+ await sync_actions.move_thread_to_folder_job({"job_try": 1}, str(ctx.thread_id), "archive")
+ assert provider.calls == 1 # stopped at the first failure, not 20 s per message
+ ctx.session.expunge_all()
+ async with session_scope() as verify:
+ inbox = await verify.scalar(
+ select(Mailbox).where(Mailbox.user_id == ctx.user_id, Mailbox.special_use == "inbox")
+ )
+ assert inbox is not None
+ moved = await verify.scalar(
+ select(func.count())
+ .select_from(Message)
+ .where(Message.thread_id == ctx.thread_id, Message.mailbox_id != inbox.id)
+ )
+ assert moved == 0 # nothing recorded as moved
+
+
+async def test_delete_keeps_rows_while_unreachable(ctx: Ctx, monkeypatch) -> None: # type: ignore[no-untyped-def]
+ monkeypatch.setattr(sync_actions, "build_provider", lambda _a: _Unreachable())
+ with pytest.raises(Retry):
+ await sync_actions.delete_thread_job({"job_try": 1}, str(ctx.thread_id))
+ ctx.session.expunge_all()
+ async with session_scope() as verify:
+ left = await verify.scalar(
+ select(func.count()).select_from(Message).where(Message.thread_id == ctx.thread_id)
+ )
+ assert left == len(ctx.message_ids) # kept: the server still has them
diff --git a/backend/tests/outbox/test_outbox.py b/backend/tests/outbox/test_outbox.py
index 7dd95ee..6831a1d 100644
--- a/backend/tests/outbox/test_outbox.py
+++ b/backend/tests/outbox/test_outbox.py
@@ -2,11 +2,12 @@
from __future__ import annotations
+import asyncio
import email
import uuid
from collections.abc import AsyncIterator
from dataclasses import dataclass, field
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
from email.policy import default as default_policy
import httpx
@@ -25,6 +26,7 @@ from app.main import create_app
from app.models import Account, Draft, KnownRecipient, Mailbox, Message, Outbox, User
from app.outbox import jobs as outbox_jobs
from app.outbox.mime_build import build_mime
+from app.providers.base import ProviderConnectError, ProviderRejectedError
CSRF = "csrf"
@@ -193,6 +195,8 @@ async def ctx(monkeypatch) -> AsyncIterator[Ctx]: # type: ignore[no-untyped-def
yield c
finally:
await c.client.aclose()
+ # outbox.user_id has no foreign key: drop the rows with the user.
+ await session.execute(delete(Outbox).where(Outbox.user_id == user.id))
await session.execute(delete(User).where(User.id == user.id))
await session.commit()
await session.close()
@@ -425,13 +429,19 @@ class FakeProvider:
self.sent: list = []
self.appended: list = []
self.raise_on_send = False
+ self.send_error: Exception | None = None
+ self.append_error: Exception | None = None
async def send(self, raw: bytes, env_from: str, env_to: list) -> None:
if self.raise_on_send:
raise RuntimeError("smtp down")
+ if self.send_error is not None:
+ raise self.send_error
self.sent.append((env_from, env_to))
async def append(self, mailbox: str, raw: bytes, flags: frozenset) -> int:
+ if self.append_error is not None:
+ raise self.append_error
self.appended.append(mailbox)
return 99
@@ -516,3 +526,124 @@ async def test_send_job_failure_marks_failed(ctx: Ctx, monkeypatch) -> None: #
ob = await ctx.session.get(Outbox, oid)
await ctx.session.refresh(ob)
assert ob is not None and ob.state == "failed" and ob.error
+
+
+# --- a send the server could not take: Scheduled, retried for 24 hours ------------
+
+
+async def _outbox(ctx: Ctx, oid: uuid.UUID) -> Outbox:
+ ob = await ctx.session.get(Outbox, oid)
+ assert ob is not None
+ await ctx.session.refresh(ob)
+ return ob
+
+
+async def test_unreachable_smtp_is_retried_and_waits_in_scheduled(ctx: Ctx, monkeypatch) -> None: # type: ignore[no-untyped-def]
+ oid = await _make_outbox(ctx)
+ fake = FakeProvider()
+ fake.send_error = ProviderConnectError("cannot connect to smtp.example.com")
+ monkeypatch.setattr(outbox_jobs, "build_provider", lambda _a: fake)
+
+ before = datetime.now(UTC)
+ await outbox_jobs.send_job({"job_try": 1}, str(oid))
+
+ ob = await _outbox(ctx, oid)
+ assert ob.state == "retrying" and ob.attempts == 1 and "cannot connect" in (ob.error or "")
+ assert ob.retry_until is not None
+ assert ob.retry_until >= before + timedelta(hours=23, minutes=59)
+ assert ob.next_attempt_at is not None
+ assert timedelta(seconds=55) < ob.next_attempt_at - before < timedelta(seconds=70)
+ args, kwargs = ctx.enqueued[-1]
+ assert args == ("send_job", str(oid))
+ assert kwargs["_defer_until"] == ob.next_attempt_at and kwargs["_job_id"] == f"send:{oid}:1"
+
+ # Waiting in Scheduled, not in Drafts.
+ drafts = (await ctx.client.get("/api/drafts")).json()
+ assert all(d["id"] != str(ob.draft_id) for d in drafts)
+ waiting = (await ctx.client.get("/api/outbox")).json()
+ assert [(w["outbox_id"], w["state"], w["subject"]) for w in waiting] == [
+ (str(oid), "retrying", "Re: Hi")
+ ]
+
+ # The server is back: the planned retry sends it.
+ fake.send_error = None
+ await outbox_jobs.send_job({"job_try": 1}, str(oid))
+ assert (await _outbox(ctx, oid)).state == "sent" and len(fake.sent) == 1
+ assert (await ctx.client.get("/api/outbox")).json() == []
+
+
+def test_retry_gaps_grow_then_hourly() -> None:
+ assert [outbox_jobs._retry_gap(n) for n in range(1, 8)] == [60, 120, 300, 600, 1800, 3600, 3600]
+
+
+async def test_send_gives_up_after_24_hours(ctx: Ctx, monkeypatch) -> None: # type: ignore[no-untyped-def]
+ oid = await _make_outbox(ctx, state="retrying")
+ ob = await _outbox(ctx, oid)
+ ob.retry_until = datetime.now(UTC) - timedelta(seconds=1)
+ await ctx.session.commit()
+ fake = FakeProvider()
+ fake.send_error = ProviderConnectError("still down")
+ monkeypatch.setattr(outbox_jobs, "build_provider", lambda _a: fake)
+
+ await outbox_jobs.send_job({"job_try": 1}, str(oid))
+ ob = await _outbox(ctx, oid)
+ assert ob.state == "failed" and ob.next_attempt_at is None
+ assert not any(a[0] == "send_job" for a, _ in ctx.enqueued)
+
+
+async def test_permanent_refusal_fails_at_once(ctx: Ctx, monkeypatch) -> None: # type: ignore[no-untyped-def]
+ oid = await _make_outbox(ctx)
+ fake = FakeProvider()
+ fake.send_error = ProviderRejectedError("550 no such user")
+ monkeypatch.setattr(outbox_jobs, "build_provider", lambda _a: fake)
+
+ await outbox_jobs.send_job({"job_try": 1}, str(oid))
+ ob = await _outbox(ctx, oid)
+ assert ob.state == "failed" and "550" in (ob.error or "")
+ assert not any(a[0] == "send_job" for a, _ in ctx.enqueued)
+ # Shown in Scheduled as failed, with the reason.
+ waiting = (await ctx.client.get("/api/outbox")).json()
+ assert waiting[0]["state"] == "failed" and "550" in waiting[0]["error"]
+
+
+async def test_delivered_but_not_filed_counts_as_sent(ctx: Ctx, monkeypatch) -> None: # type: ignore[no-untyped-def]
+ oid = await _make_outbox(ctx)
+ fake = FakeProvider()
+ fake.append_error = ProviderConnectError("imap unreachable")
+ monkeypatch.setattr(outbox_jobs, "build_provider", lambda _a: fake)
+
+ await outbox_jobs.send_job({"job_try": 1}, str(oid))
+ assert (await _outbox(ctx, oid)).state == "sent" and len(fake.sent) == 1
+
+
+async def test_two_jobs_for_one_send_send_once(ctx: Ctx, monkeypatch) -> None: # type: ignore[no-untyped-def]
+ oid = await _make_outbox(ctx, state="retrying")
+ fake = FakeProvider()
+ monkeypatch.setattr(outbox_jobs, "build_provider", lambda _a: fake)
+
+ await asyncio.gather(
+ outbox_jobs.send_job({"job_try": 1}, str(oid)),
+ outbox_jobs.send_job({"job_try": 1}, str(oid)),
+ )
+ assert len(fake.sent) == 1
+
+
+async def test_retry_now_and_cancel(ctx: Ctx) -> None:
+ oid = await _make_outbox(ctx, state="failed")
+ resp = await ctx.client.post(f"/api/outbox/{oid}/retry", headers=_hdr())
+ assert resp.status_code == 200
+ ob = await _outbox(ctx, oid)
+ assert ob.state == "retrying"
+ assert ob.retry_until is not None and ob.retry_until > datetime.now(UTC) + timedelta(hours=23)
+ assert ctx.enqueued[-1][0] == ("send_job", str(oid))
+
+ resp = await ctx.client.post(f"/api/outbox/{oid}/cancel", headers=_hdr())
+ assert resp.status_code == 200 and resp.json()["draft_id"] == str(ob.draft_id)
+ assert (await _outbox(ctx, oid)).state == "cancelled"
+ # Back in Drafts, gone from Scheduled.
+ drafts = (await ctx.client.get("/api/drafts")).json()
+ assert any(d["id"] == str(ob.draft_id) for d in drafts)
+ assert (await ctx.client.get("/api/outbox")).json() == []
+ # Only a waiting send can be retried or cancelled.
+ again = await ctx.client.post(f"/api/outbox/{oid}/retry", headers=_hdr())
+ assert again.status_code == 409
diff --git a/backend/tests/sync/test_acceptance.py b/backend/tests/sync/test_acceptance.py
index 993c9b8..a80a902 100644
--- a/backend/tests/sync/test_acceptance.py
+++ b/backend/tests/sync/test_acceptance.py
@@ -44,7 +44,9 @@ async def connections(monkeypatch) -> AsyncIterator[dict[uuid.UUID, int]]: # ty
"""Patch the runner's loops; yield a live (hot-tier IDLE) connection count."""
conns: dict[uuid.UUID, int] = defaultdict(int)
- async def fake_live_loop(account_id: uuid.UUID, folder: str, stop: asyncio.Event) -> None:
+ async def fake_live_loop(
+ account_id: uuid.UUID, folder: str, stop: asyncio.Event, health: object = None
+ ) -> None:
conns[account_id] += 1 # one persistent IDLE connection while alive (hot)
try:
await stop.wait()
@@ -52,13 +54,19 @@ async def connections(monkeypatch) -> AsyncIterator[dict[uuid.UUID, int]]: # ty
conns[account_id] -= 1
async def fake_secondary(
- 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: object = None,
) -> None:
assert stop is not None
await stop.wait()
async def fake_poll_all(
- account_id: uuid.UUID, interval: float, stop: asyncio.Event | None = None
+ account_id: uuid.UUID,
+ interval: float,
+ stop: asyncio.Event | None = None,
+ health: object = None,
) -> None:
assert stop is not None # warm/cold: no persistent connection, just wait
await stop.wait()
diff --git a/backend/tests/sync/test_account_health.py b/backend/tests/sync/test_account_health.py
new file mode 100644
index 0000000..88150ea
--- /dev/null
+++ b/backend/tests/sync/test_account_health.py
@@ -0,0 +1,127 @@
+"""An account whose mail server is unreachable, or refuses the password."""
+
+from __future__ import annotations
+
+import uuid
+from collections.abc import AsyncIterator
+
+import aiosmtplib
+import pytest_asyncio
+from aioimaplib import aioimaplib
+from sqlalchemy import delete
+
+from app.db import session_scope
+from app.identity.passwords import hash_password
+from app.models import Account, User
+from app.providers.base import (
+ ProviderAuthError,
+ ProviderConnectError,
+ ProviderRejectedError,
+)
+from app.providers.imap_smtp.provider import _smtp_error
+from app.sync.health import AUTH_GIVE_UP_AFTER, UNREACHABLE_AFTER, AccountHealth, is_temporary
+
+
+@pytest_asyncio.fixture
+async def account_id() -> AsyncIterator[uuid.UUID]:
+ async with session_scope() as session:
+ user = User(
+ handle=f"hl{uuid.uuid4().hex[:8]}@test.local", password_hash=hash_password("pw")
+ )
+ session.add(user)
+ await session.flush()
+ account = Account(
+ user_id=user.id,
+ email="me@example.com",
+ imap_config={"host": "imap.example.com", "port": 993, "tls": "ssl"},
+ smtp_config={"host": "smtp.example.com", "port": 465, "tls": "ssl"},
+ secret_ct=b"",
+ discovery_src="manual",
+ state="live",
+ )
+ session.add(account)
+ await session.flush()
+ user_id, aid = user.id, account.id
+ try:
+ yield aid
+ finally:
+ async with session_scope() as session:
+ await session.execute(delete(User).where(User.id == user_id))
+
+
+async def _account(aid: uuid.UUID) -> Account:
+ async with session_scope() as session:
+ account = await session.get(Account, aid)
+ assert account is not None
+ return account
+
+
+def test_what_counts_as_temporary() -> None:
+ assert is_temporary(ProviderConnectError("cannot connect"))
+ assert is_temporary(TimeoutError())
+ assert is_temporary(ConnectionResetError())
+ assert is_temporary(aioimaplib.CommandTimeout("uid move"))
+ # After the connection drops, aioimaplib refuses commands this way.
+ assert is_temporary(aioimaplib.Abort("command UID illegal in state LOGOUT"))
+ assert not is_temporary(aioimaplib.Abort("server has not MOVE capability"))
+ assert not is_temporary(ProviderAuthError("bad password"))
+ assert not is_temporary(ProviderRejectedError("550 no such user"))
+ assert not is_temporary(ValueError("bug"))
+
+
+def test_smtp_errors_say_whether_to_retry() -> None:
+ assert isinstance(_smtp_error(aiosmtplib.SMTPConnectError("refused")), ProviderConnectError)
+ assert isinstance(_smtp_error(aiosmtplib.SMTPServerDisconnected("gone")), ProviderConnectError)
+ assert isinstance(
+ _smtp_error(aiosmtplib.SMTPResponseException(451, "try later")), ProviderConnectError
+ )
+ assert isinstance(
+ _smtp_error(aiosmtplib.SMTPResponseException(554, "rejected")), ProviderRejectedError
+ )
+ assert isinstance(
+ _smtp_error(aiosmtplib.SMTPAuthenticationError(535, "bad credentials")), ProviderAuthError
+ )
+ refused = aiosmtplib.SMTPRecipientsRefused(
+ [aiosmtplib.SMTPRecipientRefused(550, "no such user", "x@example.com")]
+ )
+ assert isinstance(_smtp_error(refused), ProviderRejectedError)
+ greylisted = aiosmtplib.SMTPRecipientsRefused(
+ [aiosmtplib.SMTPRecipientRefused(450, "greylisted", "x@example.com")]
+ )
+ assert isinstance(_smtp_error(greylisted), ProviderConnectError)
+
+
+async def test_unreachable_after_failures_in_a_row_then_clears(account_id: uuid.UUID) -> None:
+ health = AccountHealth(account_id)
+ for _ in range(UNREACHABLE_AFTER - 1):
+ await health.failed(ProviderConnectError("cannot connect"))
+ assert (await _account(account_id)).sync_error is None # a blip is not an outage
+
+ await health.failed(ProviderConnectError("cannot connect"))
+ account = await _account(account_id)
+ assert account.sync_error == "unreachable" and account.sync_error_since is not None
+ assert account.state == "live" # still synced: retried until it answers
+
+ await health.ok()
+ account = await _account(account_id)
+ assert account.sync_error is None and account.sync_error_since is None
+
+
+async def test_mark_survives_a_restart_and_clears_on_success(account_id: uuid.UUID) -> None:
+ first = AccountHealth(account_id)
+ for _ in range(UNREACHABLE_AFTER):
+ await first.failed(ProviderConnectError("cannot connect"))
+ # A new worker knows nothing of the mark, yet its first success clears it.
+ await AccountHealth(account_id).ok()
+ assert (await _account(account_id)).sync_error is None
+
+
+async def test_refused_sign_in_stops_syncing(account_id: uuid.UUID) -> None:
+ health = AccountHealth(account_id)
+ for _ in range(AUTH_GIVE_UP_AFTER - 1):
+ await health.failed(ProviderAuthError("AUTHENTICATIONFAILED"))
+ assert (await _account(account_id)).state == "live"
+
+ await health.failed(ProviderAuthError("AUTHENTICATIONFAILED"))
+ account = await _account(account_id)
+ assert account.state == "auth_failed" and "AUTHENTICATIONFAILED" in account.state_detail
diff --git a/backend/tests/sync/test_backfill_window.py b/backend/tests/sync/test_backfill_window.py
index cbed98f..66e72a9 100644
--- a/backend/tests/sync/test_backfill_window.py
+++ b/backend/tests/sync/test_backfill_window.py
@@ -8,7 +8,7 @@ from collections.abc import AsyncIterator
from datetime import UTC, date, datetime, timedelta
from email.message import EmailMessage
-from sqlalchemy import func, select
+from sqlalchemy import delete, func, select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.pool import NullPool
@@ -16,7 +16,9 @@ from app.config import get_settings
from app.identity.passwords import hash_password
from app.models import Account, Mailbox, Message, User
from app.providers.base import RemoteMailbox, RemoteMessage
+from app.redis_client import get_redis
from app.sync import engine as sync_engine
+from app.sync import leases
def _raw(mid: str) -> bytes:
@@ -89,6 +91,16 @@ async def _seed(sm: async_sessionmaker, window: str) -> tuple[uuid.UUID, uuid.UU
return user.id, aid
+async def _cleanup(sm: async_sessionmaker, user_id: uuid.UUID, account_id: uuid.UUID) -> None: # type: ignore[type-arg]
+ """Remove the seeded user — backfill sets its account live, and left behind a
+ running dev worker would keep trying to sync it."""
+ await leases.mark_not_live(get_redis(), account_id)
+ session = sm()
+ await session.execute(delete(User).where(User.id == user_id))
+ await session.commit()
+ await session.close()
+
+
async def test_window_new_sets_watermark_and_skips_backfill(monkeypatch) -> None: # type: ignore[no-untyped-def]
engine = create_async_engine(get_settings().database_url, poolclass=NullPool)
sm = async_sessionmaker(engine, expire_on_commit=False)
@@ -112,6 +124,7 @@ async def test_window_new_sets_watermark_and_skips_backfill(monkeypatch) -> None
assert msgs == 0 # nothing imported
await session.close()
finally:
+ await _cleanup(sm, _uid, account_id)
await engine.dispose()
@@ -142,6 +155,7 @@ async def test_window_all_imports_recent_then_remainder(monkeypatch) -> None: #
assert msgs == 2 # one from each pass
await session.close()
finally:
+ await _cleanup(sm, _uid, account_id)
await engine.dispose()
@@ -157,6 +171,6 @@ async def test_window_3months_single_dated_pass(monkeypatch) -> None: # type: i
since = date.today() - timedelta(days=sync_engine._WINDOW_DAYS["3months"])
assert fake.backfill_calls == [("INBOX", since, None)] # single windowed pass
assert fake.count_calls == [("INBOX", since)] # denominator counts the window
- await engine.dispose()
finally:
- pass
+ await _cleanup(sm, _uid, account_id)
+ await engine.dispose()
diff --git a/backend/workers/worker.py b/backend/workers/worker.py
index 0724dc7..ea55937 100644
--- a/backend/workers/worker.py
+++ b/backend/workers/worker.py
@@ -12,9 +12,12 @@ import logging
import uuid
from typing import Any
+from arq import func
+
from app.outbox.draft_sync import push_draft_to_imap_job, remove_imap_draft_job
from app.outbox.jobs import send_job
from app.sync.actions import (
+ ACTION_MAX_TRIES,
delete_thread_job,
empty_folder_job,
move_job,
@@ -35,14 +38,20 @@ async def sync_bootstrap_account(ctx: dict[str, Any], account_id: str) -> int:
return count
-# Fast, interactive user actions (§6): short timeout, high concurrency.
-INTERACTIVE_FUNCTIONS = [
+# Fast, interactive user actions (§6): short timeout, high concurrency. The ones
+# pushing a change to the mail server retry (arq ``Retry``) for ~24 hours while
+# the server is unreachable — more tries than arq's default of 5 allows.
+_RETRYING_ACTIONS: list[Any] = [
push_flags_job,
move_job,
move_thread_to_folder_job,
restore_thread_job,
delete_thread_job,
empty_folder_job,
+]
+
+INTERACTIVE_FUNCTIONS: list[Any] = [
+ *(func(job, max_tries=ACTION_MAX_TRIES) for job in _RETRYING_ACTIONS),
send_job,
push_draft_to_imap_job,
remove_imap_draft_job,
diff --git a/frontend/src/components/Offline.tsx b/frontend/src/components/Offline.tsx
index 2b6a17d..074c750 100644
--- a/frontend/src/components/Offline.tsx
+++ b/frontend/src/components/Offline.tsx
@@ -200,24 +200,37 @@ export function OfflineController() {
role="status"
className="wm-offline-notice pointer-events-none fixed bottom-6 left-1/2 z-[70] flex max-w-[calc(100vw-2rem)] -translate-x-1/2 items-center gap-2.5 rounded-xl border border-app-border bg-app-elevated px-4 py-2.5 text-sm text-app-fg shadow-lg"
>
- {current.key === "scheduled" || current.key === "sendingScheduled" ? (
+ {current.key === "scheduledSent" ? (
+
+