From 23704f323c3968c1a31672c1995a44a9113c3257 Mon Sep 17 00:00:00 2001 From: Lukas Remis Date: Sat, 18 Jul 2026 19:56:06 +0200 Subject: [PATCH] sync-v2 step 2: extract AccountRunner Move the per-account INBOX-IDLE + secondary-poll loops out of the supervisor into a self-contained AccountRunner (start/stop/teardown). The startup supervisor now launches one runner per live account. Pure refactor, no behavior change. Co-Authored-By: Claude Opus 4.8 --- backend/app/sync/runner.py | 68 ++++++++++++++++++++++++++++++ backend/app/sync/supervisor.py | 75 ++++++++++------------------------ 2 files changed, 90 insertions(+), 53 deletions(-) create mode 100644 backend/app/sync/runner.py diff --git a/backend/app/sync/runner.py b/backend/app/sync/runner.py new file mode 100644 index 0000000..220fb26 --- /dev/null +++ b/backend/app/sync/runner.py @@ -0,0 +1,68 @@ +"""Per-account sync runner (refactor §4). + +An ``AccountRunner`` owns the long-lived sync loops for a single account and +exposes a small ``start`` / ``stop`` lifecycle. This is a pure extraction of the +loops that previously lived inline in the supervisor — behavior is unchanged +(INBOX IDLE + secondary CONDSTORE poll, each self-healing with backoff). +""" + +from __future__ import annotations + +import asyncio +import contextlib +import logging +import uuid +from collections.abc import Awaitable, Callable + +from app.sync.engine import live_loop, secondary_poll_loop + +logger = logging.getLogger(__name__) + +_RECONNECT_BACKOFF = 30.0 + + +class AccountRunner: + """Owns the sync loops for one account until stopped.""" + + def __init__(self, account_id: uuid.UUID) -> None: + self.account_id = account_id + self._stop = asyncio.Event() + self._tasks: list[asyncio.Task[None]] = [] + + def start(self) -> None: + """Launch the account's sync loops (idempotent-ish: call once).""" + self._stop.clear() + self._tasks = [ + asyncio.create_task(self._supervise(self._run_live, "live")), + asyncio.create_task(self._supervise(self._run_secondary, "secondary")), + ] + + def running(self) -> bool: + return any(not t.done() for t in self._tasks) + + async def stop(self) -> None: + """Signal and cancel the loops, waiting for teardown.""" + self._stop.set() + for task in self._tasks: + task.cancel() + if self._tasks: + await asyncio.gather(*self._tasks, return_exceptions=True) + self._tasks = [] + + async def _run_live(self) -> None: + await live_loop(self.account_id, "INBOX", self._stop) + + async def _run_secondary(self) -> None: + await secondary_poll_loop(self.account_id, self._stop) + + async def _supervise(self, run: Callable[[], Awaitable[None]], label: str) -> None: + """Run a loop, restarting it (with backoff) until stopped.""" + while not self._stop.is_set(): + try: + await run() + except asyncio.CancelledError: + raise + except Exception: + logger.exception("%s loop for account %s crashed; retrying", label, self.account_id) + with contextlib.suppress(TimeoutError): + await asyncio.wait_for(self._stop.wait(), timeout=_RECONNECT_BACKOFF) diff --git a/backend/app/sync/supervisor.py b/backend/app/sync/supervisor.py index 93bbeec..357bcd3 100644 --- a/backend/app/sync/supervisor.py +++ b/backend/app/sync/supervisor.py @@ -1,17 +1,14 @@ -"""Live-sync supervisor (SPEC §6.3). +"""Live-sync supervisor (refactor §4, transitional). -Runs inside the arq worker: on startup it launches one IDLE/poll loop per live -account so new mail and flag changes flow to the UI over WebSocket. Loops -self-heal (reconnect with backoff) and are cancelled on worker shutdown. - -NOTE(spec): accounts that become live *after* worker startup are picked up on the -next worker restart; reacting to ``account.state_changed`` live is a later refinement. +Runs inside the sync worker: on startup it launches one :class:`AccountRunner` +per live account. This step (v2 §9.2) only extracts the per-account loops into a +runner; account discovery is still startup-only and is replaced by the lease +claimer in a later step. """ from __future__ import annotations import asyncio -import contextlib import logging import uuid from typing import Any @@ -20,58 +17,30 @@ from sqlalchemy import select from app.db import session_scope from app.models import Account -from app.sync.engine import live_loop, secondary_poll_loop +from app.sync.runner import AccountRunner logger = logging.getLogger(__name__) -_RECONNECT_BACKOFF = 30.0 - -async def _run_account(account_id: uuid.UUID, stop: asyncio.Event) -> None: - while not stop.is_set(): - try: - await live_loop(account_id, "INBOX", stop) - except asyncio.CancelledError: - raise - except Exception: - logger.exception("live loop for account %s crashed; reconnecting", account_id) - with contextlib.suppress(TimeoutError): - await asyncio.wait_for(stop.wait(), timeout=_RECONNECT_BACKOFF) - - -async def _run_secondary(account_id: uuid.UUID, stop: asyncio.Event) -> None: - while not stop.is_set(): - try: - await secondary_poll_loop(account_id, stop) - except asyncio.CancelledError: - raise - except Exception: - logger.exception("secondary poll loop for %s crashed; retrying", account_id) - with contextlib.suppress(TimeoutError): - await asyncio.wait_for(stop.wait(), timeout=_RECONNECT_BACKOFF) +async def _live_account_ids() -> list[uuid.UUID]: + async with session_scope() as session: + return list(await session.scalars(select(Account.id).where(Account.state == "live"))) async def start_supervisor(ctx: dict[str, Any]) -> None: - """Launch a live loop per live account (called from the worker on_startup).""" - async with session_scope() as session: - account_ids = list(await session.scalars(select(Account.id).where(Account.state == "live"))) - stop = asyncio.Event() - ctx["supervisor_stop"] = stop - tasks: list[asyncio.Task[None]] = [] - for aid in account_ids: - tasks.append(asyncio.create_task(_run_account(aid, stop))) # INBOX IDLE - tasks.append(asyncio.create_task(_run_secondary(aid, stop))) # other folders poll - ctx["supervisor_tasks"] = tasks - logger.info("live supervisor started for %d account(s)", len(account_ids)) + """Launch an AccountRunner per live account (called from the worker on_startup).""" + runners: dict[uuid.UUID, AccountRunner] = {} + for account_id in await _live_account_ids(): + runner = AccountRunner(account_id) + runner.start() + runners[account_id] = runner + ctx["runners"] = runners + logger.info("live supervisor started for %d account(s)", len(runners)) async def stop_supervisor(ctx: dict[str, Any]) -> None: - """Signal and cancel all live loops (called from the worker on_shutdown).""" - stop = ctx.get("supervisor_stop") - if stop is not None: - stop.set() - tasks = ctx.get("supervisor_tasks", []) - for task in tasks: - task.cancel() - if tasks: - await asyncio.gather(*tasks, return_exceptions=True) + """Stop and tear down all account runners (called from the worker on_shutdown).""" + runners: dict[uuid.UUID, AccountRunner] = ctx.get("runners", {}) + if runners: + await asyncio.gather(*(r.stop() for r in runners.values()), return_exceptions=True) + ctx["runners"] = {}