"""Per-account sync runner (refactor §4, §5). An ``AccountRunner`` owns the long-lived sync work for a single account and exposes a ``start`` / ``stop`` / ``apply_tier`` lifecycle. The task set is shaped by the account's activity tier (§5): * **hot** — a persistent INBOX IDLE loop plus a secondary CONDSTORE poll of the special folders every ``POLL_WARM`` seconds. * **warm** — no persistent socket; a single transient poll of INBOX + special folders every ``POLL_WARM`` seconds. * **cold** — the same single transient poll, every ``POLL_COLD`` seconds. Correctness is identical across tiers — only latency and socket cost differ. """ from __future__ import annotations import asyncio import contextlib import logging import random import uuid from collections.abc import Awaitable, Callable 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" class AccountRunner: """Owns the tier-shaped sync work for one account until stopped.""" def __init__(self, account_id: uuid.UUID, tier: Tier = "warm") -> None: self.account_id = account_id self.tier = tier self._stop = asyncio.Event() self._tasks: list[asyncio.Task[None]] = [] # tier-independent (verify) self._work: list[asyncio.Task[None]] = [] # tier-shaped loops # Set when the account is found no longer live (§3 defense in depth); the # 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.""" self._stop.clear() self._tasks = [asyncio.create_task(self._verify_loop())] self._start_work() def running(self) -> bool: return any(not t.done() for t in (*self._tasks, *self._work)) async def stop(self) -> None: """Signal and cancel every loop, waiting for teardown.""" self._stop.set() await self._cancel(self._work) await self._cancel(self._tasks) self._work = [] self._tasks = [] async def apply_tier(self, tier: Tier) -> None: """Reshape the work loops for a new tier (no-op if unchanged).""" if tier == self.tier and self._work: return logger.info("account %s tier %s -> %s", self.account_id, self.tier, tier) self.tier = tier await self._cancel(self._work) self._start_work() # --- internals ----------------------------------------------------------- def _start_work(self) -> None: settings = get_settings() if self.tier == "hot": self._work = [ asyncio.create_task(self._supervise(self._run_live, "live")), asyncio.create_task(self._supervise(self._run_secondary, "secondary")), ] else: interval = float(settings.poll_warm if self.tier == "warm" else settings.poll_cold) self._work = [ asyncio.create_task(self._supervise(lambda: self._run_poll(interval), "poll")) ] async def _cancel(self, tasks: list[asyncio.Task[None]]) -> None: for task in tasks: task.cancel() if tasks: await asyncio.gather(*tasks, return_exceptions=True) async def _run_live(self) -> None: 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, self.health ) async def _run_poll(self, interval: float) -> None: 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. This makes zombie loops structurally impossible even if a lifecycle event is lost: an account that was disabled/deleted stops being synced within a verify cycle regardless of the lease/event paths. """ while not self._stop.is_set(): with contextlib.suppress(TimeoutError): await asyncio.wait_for(self._stop.wait(), timeout=self.verify_interval) if self._stop.is_set(): return if not await self._account_is_live(): logger.info("account %s no longer live; flagging runner dead", self.account_id) self.dead.set() return async def _account_is_live(self) -> bool: async with session_scope() as session: state = await session.scalar(select(Account.state).where(Account.id == self.account_id)) return state == "live" async def _supervise(self, run: Callable[[], Awaitable[None]], label: str) -> None: """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 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=delay)