diff --git a/backend/app/config.py b/backend/app/config.py index 24b900d..5e2075c 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -29,7 +29,6 @@ class Settings(BaseSettings): meili_key: str = Field(default="", alias="MEILI_KEY") undo_seconds: int = Field(default=10, alias="UNDO_SECONDS") - poll_interval: int = Field(default=120, alias="POLL_INTERVAL") session_ttl_days: int = Field(default=30, alias="SESSION_TTL_DAYS") max_attachment_mb: int = Field(default=25, alias="MAX_ATTACHMENT_MB") log_level: str = Field(default="INFO", alias="LOG_LEVEL") @@ -44,6 +43,13 @@ class Settings(BaseSettings): max_accounts_per_worker: int = Field(default=400, alias="MAX_ACCOUNTS_PER_WORKER") lease_ttl: int = Field(default=60, alias="LEASE_TTL") lease_renew: int = Field(default=20, alias="LEASE_RENEW") + # Activity tiers (§5): hot = persistent IDLE, warm/cold = transient polls. + poll_warm: int = Field(default=180, alias="POLL_WARM") + poll_cold: int = Field(default=900, alias="POLL_COLD") + tier_hot_activity_h: int = Field(default=24, alias="TIER_HOT_ACTIVITY_H") + tier_session_grace_min: int = Field(default=30, alias="TIER_SESSION_GRACE_MIN") + tier_cold_after_d: int = Field(default=3, alias="TIER_COLD_AFTER_D") + hot_budget_per_worker: int = Field(default=250, alias="HOT_BUDGET_PER_WORKER") oauth_google_client_id: str = Field(default="", alias="OAUTH_GOOGLE_CLIENT_ID") oauth_google_client_secret: str = Field(default="", alias="OAUTH_GOOGLE_CLIENT_SECRET") diff --git a/backend/app/identity/routes.py b/backend/app/identity/routes.py index 353cf54..6843934 100644 --- a/backend/app/identity/routes.py +++ b/backend/app/identity/routes.py @@ -51,6 +51,8 @@ from app.identity.sessions import ( revoke_session, ) from app.models import User +from app.redis_client import get_redis +from app.sync import leases router = APIRouter(prefix="/auth", tags=["auth"]) @@ -117,7 +119,10 @@ async def login(body: LoginRequest, request: Request, response: Response, db: Db if user is None: raise_api_error(401, "invalid_credentials", "incorrect handle or password") await _establish_session(request, response, user) - # Sync v2 §3: a login marks the user active, promoting their accounts' tier. + # Sync v2 §3/§5: a login marks the user active (session heat) and promotes + # their accounts' tier to hot. + settings = get_settings() + await leases.mark_user_active(get_redis(), user.id, settings.tier_session_grace_min * 60) await get_event_bus().publish("user.active", {"user_id": str(user.id)}) return _user_out(user) diff --git a/backend/app/providers/base.py b/backend/app/providers/base.py index 7b2fea0..883e4d4 100644 --- a/backend/app/providers/base.py +++ b/backend/app/providers/base.py @@ -85,7 +85,7 @@ class MailProvider(Protocol): async def idle(self, mailbox: str) -> None: """Blocks until the server signals change or 25 min elapse (re-issue IDLE). May raise - ProviderNotSupported => sync loop falls back to poll() every POLL_INTERVAL (default 120 s). + ProviderNotSupported => sync loop falls back to poll() every POLL_WARM (default 180 s). """ ... diff --git a/backend/app/sync/coordinator.py b/backend/app/sync/coordinator.py index 11bcc0f..837188d 100644 --- a/backend/app/sync/coordinator.py +++ b/backend/app/sync/coordinator.py @@ -25,13 +25,14 @@ import secrets import socket import time import uuid +from datetime import UTC, datetime, timedelta -from sqlalchemy import select +from sqlalchemy import func, select from app.config import get_settings from app.db import session_scope from app.events import channel_for -from app.models import Account +from app.models import Account, Message from app.redis_client import aw, get_redis from app.sync import leases from app.sync.runner import AccountRunner @@ -65,6 +66,11 @@ class SyncCoordinator: self.heartbeat_ttl = 30 self.heartbeat_refresh = 10.0 self.rebalance_interval = 60.0 + self.tier_eval_interval = 120.0 + self.session_grace_s = s.tier_session_grace_min * 60 + self.hot_activity = timedelta(hours=s.tier_hot_activity_h) + self.cold_after = timedelta(days=s.tier_cold_after_d) + self.hot_budget = s.hot_budget_per_worker # --- lifecycle ----------------------------------------------------------- @@ -80,6 +86,7 @@ class SyncCoordinator: asyncio.create_task(self._loop(self._reconcile_once, self.reconcile_interval)), asyncio.create_task(self._loop(self._heartbeat_once, self.heartbeat_refresh)), asyncio.create_task(self._loop(self._rebalance_once, self.rebalance_interval)), + asyncio.create_task(self._loop(self._evaluate_tiers_once, self.tier_eval_interval)), asyncio.create_task(self._event_subscriber()), ] logger.info("sync coordinator %s started", self.worker_id) @@ -131,7 +138,7 @@ class SyncCoordinator: if account_id in self._runners: continue if await leases.claim(self._redis, account_id, self.worker_id, self.lease_ttl): - self._start_runner(account_id) + await self._start_runner(account_id) async def _renew_once(self) -> None: for account_id in list(self._runners): @@ -160,8 +167,11 @@ class SyncCoordinator: if runner is not None: await runner.stop() - def _start_runner(self, account_id: uuid.UUID) -> None: - runner = AccountRunner(account_id) + async def _start_runner(self, account_id: uuid.UUID) -> None: + # Inherit any tier a previous owner recorded; default warm on first claim. + tier = await aw(self._redis.get(leases.tier_key(account_id))) or "warm" + await aw(self._redis.set(leases.tier_key(account_id), tier)) + runner = AccountRunner(account_id, tier) runner.start() self._runners[account_id] = runner @@ -225,7 +235,9 @@ class SyncCoordinator: the lease to lapse. """ pubsub = self._redis.pubsub() - await pubsub.subscribe(channel_for("account.state_changed")) + state_ch = channel_for("account.state_changed") + active_ch = channel_for("user.active") + await pubsub.subscribe(state_ch, active_ch) try: async for message in pubsub.listen(): if self._stop.is_set(): @@ -236,7 +248,10 @@ class SyncCoordinator: payload = json.loads(message["data"]).get("payload", {}) except (ValueError, TypeError): continue - await self._on_state_changed(payload) + if message.get("channel") == active_ch: + await self._on_user_active(payload) + else: + await self._on_state_changed(payload) finally: with contextlib.suppress(Exception): await pubsub.aclose() # type: ignore[no-untyped-call] @@ -250,3 +265,84 @@ class SyncCoordinator: await leases.mark_live(self._redis, account_id) # claimable next cycle elif account_id in self._runners: await self._release(account_id) # disabled/deleted/error -> stop now + + async def _on_user_active(self, payload: dict[str, object]) -> None: + """Promote all this user's owned accounts to hot at once (≤5s, §5).""" + raw = payload.get("user_id") + if not isinstance(raw, str): + return + user_id = uuid.UUID(raw) + owned = list(self._runners) + if not owned: + return + async with session_scope() as session: + aids = list( + await session.scalars( + select(Account.id).where(Account.id.in_(owned), Account.user_id == user_id) + ) + ) + for account_id in aids: + await self._set_tier(account_id, "hot") + + # --- tier evaluation (§5) ------------------------------------------------ + + async def _set_tier(self, account_id: uuid.UUID, tier: str) -> None: + runner = self._runners.get(account_id) + if runner is None: + return + await aw(self._redis.set(leases.tier_key(account_id), tier)) + await runner.apply_tier(tier) + + async def _account_activity( + self, ids: list[uuid.UUID] + ) -> dict[uuid.UUID, tuple[uuid.UUID, datetime | None]]: + """Return {account_id: (user_id, last_mail_at)} for the owned accounts.""" + async with session_scope() as session: + rows = ( + await session.execute( + select(Account.id, Account.user_id, func.max(Message.internal_date)) + .select_from(Account) + .outerjoin(Message, Message.account_id == Account.id) + .where(Account.id.in_(ids)) + .group_by(Account.id, Account.user_id) + ) + ).all() + return {row[0]: (row[1], row[2]) for row in rows} + + async def _evaluate_tiers_once(self) -> None: + """Recompute each owned account's tier from session heat + recent mail.""" + owned = list(self._runners) + if not owned: + return + activity = await self._account_activity(owned) + now = datetime.now(UTC) + sessions: dict[uuid.UUID, bool] = {} + desired: dict[uuid.UUID, str] = {} + demotable: list[tuple[datetime, uuid.UUID]] = [] # activity-hot, no session + hot_count = 0 + for account_id in owned: + user_id, last_mail = activity.get(account_id, (None, None)) + session_active = False + if user_id is not None: + if user_id not in sessions: + sessions[user_id] = bool( + await aw(self._redis.exists(leases.session_key(user_id))) + ) + session_active = sessions[user_id] + recent = last_mail is not None and (now - last_mail) < self.hot_activity + if session_active or recent: + desired[account_id] = "hot" + hot_count += 1 + if not session_active and last_mail is not None: + demotable.append((last_mail, account_id)) + elif last_mail is not None and (now - last_mail) < self.cold_after: + desired[account_id] = "warm" + else: + desired[account_id] = "cold" + # Global hot budget: shed the least-recently-active session-less hot accounts. + if hot_count > self.hot_budget: + demotable.sort(key=lambda item: item[0]) # oldest activity first + for _, account_id in demotable[: hot_count - self.hot_budget]: + desired[account_id] = "warm" + for account_id, tier in desired.items(): + await self._set_tier(account_id, tier) diff --git a/backend/app/sync/engine.py b/backend/app/sync/engine.py index 68b1e84..3bcf260 100644 --- a/backend/app/sync/engine.py +++ b/backend/app/sync/engine.py @@ -255,7 +255,7 @@ async def live_loop( from app.db import session_scope stop = stop or asyncio.Event() - poll_interval = get_settings().poll_interval + idle_fallback = get_settings().poll_warm async with session_scope() as session: account = await _load_account(session, account_id) @@ -267,7 +267,7 @@ async def live_loop( try: await provider.idle(mailbox_path) except ProviderNotSupported: - await asyncio.sleep(poll_interval) + await asyncio.sleep(idle_fallback) if stop.is_set(): break async with session_scope() as session: @@ -341,19 +341,39 @@ async def refresh_account_job(ctx: dict[str, Any], account_id: str) -> None: await poll_folders_once(uuid.UUID(account_id), include_inbox=True) -async def secondary_poll_loop(account_id: uuid.UUID, stop: asyncio.Event | None = None) -> None: +async def secondary_poll_loop( + account_id: uuid.UUID, stop: asyncio.Event | None = None, interval: float | None = None +) -> None: """Periodically CONDSTORE-poll non-INBOX folders (Sent/Drafts/Archive/Trash/Junk). INBOX has its own IDLE loop (real-time). This keeps the other folders in step with server-side changes (e.g. Gmail-filed sent mail) and changes made by other - clients. CONDSTORE/QRESYNC makes each pass a cheap MODSEQ delta. + clients. CONDSTORE/QRESYNC makes each pass a cheap MODSEQ delta. Used by hot-tier + runners alongside the INBOX IDLE loop. """ stop = stop or asyncio.Event() - interval = get_settings().poll_interval + period = interval if interval is not None else get_settings().poll_warm while not stop.is_set(): # Wait first: the initial state was already covered by the backfill. + with contextlib.suppress(TimeoutError): + await asyncio.wait_for(stop.wait(), timeout=period) + if stop.is_set(): + break + await _poll_secondary_once(account_id) + + +async def poll_all_loop( + account_id: uuid.UUID, interval: float, stop: asyncio.Event | None = None +) -> None: + """Transient CONDSTORE poll of INBOX + special folders every ``interval`` seconds. + + The warm/cold tiers use this instead of a persistent IDLE connection: no socket + is held between passes. Correctness matches the hot tier — only latency differs. + """ + stop = stop or asyncio.Event() + while not stop.is_set(): with contextlib.suppress(TimeoutError): await asyncio.wait_for(stop.wait(), timeout=interval) if stop.is_set(): break - await _poll_secondary_once(account_id) + await poll_folders_once(account_id, include_inbox=True) diff --git a/backend/app/sync/leases.py b/backend/app/sync/leases.py index 2918bfe..80bf693 100644 --- a/backend/app/sync/leases.py +++ b/backend/app/sync/leases.py @@ -20,6 +20,8 @@ from app.redis_client import aw LIVE_SET = "sync:live" _LEASE_PREFIX = "sync:lease:" WORKERS_PREFIX = "sync:workers:" +_TIER_PREFIX = "sync:tier:" +_SESSION_PREFIX = "sess:active:" def lease_key(account_id: uuid.UUID) -> str: @@ -30,6 +32,20 @@ def worker_key(worker_id: str) -> str: return f"{WORKERS_PREFIX}{worker_id}" +def tier_key(account_id: uuid.UUID) -> str: + return f"{_TIER_PREFIX}{account_id}" + + +def session_key(user_id: uuid.UUID | str) -> str: + """Presence key the API refreshes while a user has an active session (§5).""" + return f"{_SESSION_PREFIX}{user_id}" + + +async def mark_user_active(redis: aioredis.Redis, user_id: uuid.UUID | str, grace_s: int) -> None: + """Refresh the user's session-heat key; workers read it for tier promotion.""" + await aw(redis.set(session_key(user_id), "1", ex=grace_s)) + + # Renew only if we still own the lease (guards against a stale worker extending a # lease another worker has already claimed after expiry). Returns 1 on success. _RENEW_LUA = """ diff --git a/backend/app/sync/runner.py b/backend/app/sync/runner.py index c06977b..17c4bef 100644 --- a/backend/app/sync/runner.py +++ b/backend/app/sync/runner.py @@ -1,9 +1,16 @@ -"""Per-account sync runner (refactor §4). +"""Per-account sync runner (refactor §4, §5). -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). +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 @@ -16,54 +23,88 @@ 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.sync.engine import live_loop, secondary_poll_loop +from app.sync.engine import live_loop, poll_all_loop, secondary_poll_loop logger = logging.getLogger(__name__) _RECONNECT_BACKOFF = 30.0 _VERIFY_INTERVAL = 60.0 +Tier = str # "hot" | "warm" | "cold" + class AccountRunner: - """Owns the sync loops for one account until stopped.""" + """Owns the tier-shaped sync work for one account until stopped.""" - def __init__(self, account_id: uuid.UUID) -> None: + 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]] = [] + 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 def start(self) -> None: - """Launch the account's sync loops (idempotent-ish: call once).""" + """Launch the verify task and the current tier's work loops.""" self._stop.clear() - self._tasks = [ - asyncio.create_task(self._supervise(self._run_live, "live")), - asyncio.create_task(self._supervise(self._run_secondary, "secondary")), - asyncio.create_task(self._verify_loop()), - ] + 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) + return any(not t.done() for t in (*self._tasks, *self._work)) async def stop(self) -> None: - """Signal and cancel the loops, waiting for teardown.""" + """Signal and cancel every loop, waiting for teardown.""" self._stop.set() - for task in self._tasks: - task.cancel() - if self._tasks: - await asyncio.gather(*self._tasks, return_exceptions=True) + 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) async def _run_secondary(self) -> None: - await secondary_poll_loop(self.account_id, self._stop) + await secondary_poll_loop(self.account_id, self._stop, get_settings().poll_warm) + + async def _run_poll(self, interval: float) -> None: + await poll_all_loop(self.account_id, interval, self._stop) async def _verify_loop(self) -> None: """Periodically confirm the account is still live; flag ``dead`` if not. @@ -88,7 +129,7 @@ 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.""" + """Run a loop, restarting it (with backoff) until stopped or cancelled.""" while not self._stop.is_set(): try: await run() diff --git a/backend/app/ws.py b/backend/app/ws.py index dc58749..85e0e42 100644 --- a/backend/app/ws.py +++ b/backend/app/ws.py @@ -13,10 +13,12 @@ import logging from fastapi import APIRouter, WebSocket, WebSocketDisconnect +from app.config import get_settings from app.events import get_event_bus from app.identity.cookies import SESSION_COOKIE_INSECURE, SESSION_COOKIE_SECURE from app.identity.sessions import resolve_session from app.redis_client import get_redis +from app.sync import leases logger = logging.getLogger(__name__) @@ -35,7 +37,9 @@ async def ws_endpoint(websocket: WebSocket) -> None: await websocket.accept() user_id = str(session.user_id) - # Sync v2 §3: an open WS means the user is active -> promote their tier. + grace_s = get_settings().tier_session_grace_min * 60 + # Sync v2 §3/§5: an open WS means the user is active -> session heat + promote. + await leases.mark_user_active(get_redis(), user_id, grace_s) await get_event_bus().publish("user.active", {"user_id": user_id}) pubsub = get_redis().pubsub() await pubsub.psubscribe("evt:*") @@ -57,6 +61,8 @@ async def ws_endpoint(websocket: WebSocket) -> None: raw = await websocket.receive_text() try: if json.loads(raw).get("type") == "ping": + # A ping keeps the user's session heat alive (§5). + await leases.mark_user_active(get_redis(), user_id, grace_s) await websocket.send_json({"type": "pong"}) except (ValueError, TypeError): continue diff --git a/backend/tests/sync/test_acceptance.py b/backend/tests/sync/test_acceptance.py index cd02c8b..a661c9b 100644 --- a/backend/tests/sync/test_acceptance.py +++ b/backend/tests/sync/test_acceptance.py @@ -19,16 +19,18 @@ import asyncio import uuid from collections import defaultdict from collections.abc import AsyncIterator, Callable +from datetime import UTC, datetime, timedelta import pytest_asyncio -from sqlalchemy import delete +from sqlalchemy import delete, select from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from sqlalchemy.pool import NullPool from app.config import get_settings +from app.db import session_scope from app.events import get_event_bus from app.identity.passwords import hash_password -from app.models import Account, User +from app.models import Account, Mailbox, Message, User from app.redis_client import get_redis from app.sync import coordinator as coordinator_mod from app.sync import leases, runner @@ -39,22 +41,31 @@ from app.sync.coordinator import SyncCoordinator @pytest_asyncio.fixture async def connections(monkeypatch) -> AsyncIterator[dict[uuid.UUID, int]]: # type: ignore[no-untyped-def] - """Patch the runner's loops; yield a live-connection count per account.""" + """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: - conns[account_id] += 1 # one IDLE connection while the loop is alive + conns[account_id] += 1 # one persistent IDLE connection while alive (hot) try: await stop.wait() finally: conns[account_id] -= 1 - async def fake_secondary(account_id: uuid.UUID, stop: asyncio.Event | None = None) -> None: + async def fake_secondary( + account_id: uuid.UUID, stop: asyncio.Event | None = None, interval: float | None = 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 + ) -> None: + assert stop is not None # warm/cold: no persistent connection, just wait + await stop.wait() + monkeypatch.setattr(runner, "live_loop", fake_live_loop) monkeypatch.setattr(runner, "secondary_poll_loop", fake_secondary) + monkeypatch.setattr(runner, "poll_all_loop", fake_poll_all) yield conns @@ -119,13 +130,44 @@ def _fast(c: SyncCoordinator) -> SyncCoordinator: c.reconcile_interval = 1000.0 c.heartbeat_refresh = 1000.0 c.rebalance_interval = 1000.0 + c.tier_eval_interval = 1000.0 # lease tests pin the tier; no eval churn c._reconcile_once = _no_reconcile # type: ignore[method-assign] return c -async def _seed(ids: set[uuid.UUID]) -> None: +async def _seed(ids: set[uuid.UUID], tier: str = "hot") -> None: + """Register accounts as live and pin their tier (hot => persistent IDLE).""" + redis = get_redis() for account_id in ids: - await leases.mark_live(get_redis(), account_id) + await leases.mark_live(redis, account_id) + await redis.set(leases.tier_key(account_id), tier) + + +async def _user_of(account_id: uuid.UUID) -> uuid.UUID: + async with session_scope() as session: + user_id = await session.scalar(select(Account.user_id).where(Account.id == account_id)) + assert user_id is not None + return user_id + + +async def _add_recent_mail(account_id: uuid.UUID, when: datetime) -> None: + """Insert one INBOX message dated ``when`` so the account reads as active.""" + user_id = await _user_of(account_id) + async with session_scope() as session: + mailbox = Mailbox(account_id=account_id, user_id=user_id, path="INBOX", special_use="inbox") + session.add(mailbox) + await session.flush() + session.add( + Message( + user_id=user_id, + account_id=account_id, + mailbox_id=mailbox.id, + uid=1, + internal_date=when, + blob_sha256="0" * 64, + ) + ) + await session.commit() async def _until(pred: Callable[[], bool], timeout_s: float = 6.0) -> None: @@ -210,6 +252,7 @@ async def test_acceptance_3_create_delete(connections, make_accounts) -> None: # A newly-live account (as the bootstrap job's _set_state('live') would do): # register it in sync:live; the running coordinator must claim it, no restart. (aid,) = await make_accounts(1) + await get_redis().set(leases.tier_key(aid), "hot") # persistent IDLE await leases.mark_live(get_redis(), aid) await _until(lambda: aid in c.owned) assert connections[aid] == 1 @@ -254,5 +297,90 @@ async def test_event_driven_teardown(connections, make_accounts) -> None: # typ await c.stop() +# --- §5 activity tiers ------------------------------------------------------- + + +async def test_tier_session_promotes_to_hot(connections, make_accounts) -> None: # type: ignore[no-untyped-def] + (aid,) = await make_accounts(1) + user_id = await _user_of(aid) + await leases.mark_live(get_redis(), aid) # claimed as warm by default + await leases.mark_user_active(get_redis(), user_id, 300) # active session -> hot + c = _fast(SyncCoordinator("w1")) + c.tier_eval_interval = 0.1 + await c.start() + try: + await _until(lambda: aid in c.owned) + await _until(lambda: connections[aid] == 1) # hot => persistent IDLE + assert c._runners[aid].tier == "hot" + finally: + await c.stop() + + +async def test_tier_no_session_no_mail_goes_cold(connections, make_accounts) -> None: # type: ignore[no-untyped-def] + (aid,) = await make_accounts(1) + await leases.mark_live(get_redis(), aid) # no session, no mail + c = _fast(SyncCoordinator("w1")) + c.tier_eval_interval = 0.1 + await c.start() + try: + await _until(lambda: aid in c.owned) + await _until(lambda: c._runners[aid].tier == "cold") + assert connections[aid] == 0 # cold holds no persistent connection + finally: + await c.stop() + + +async def test_tier_recent_mail_promotes_to_hot(connections, make_accounts) -> None: # type: ignore[no-untyped-def] + (aid,) = await make_accounts(1) + await _add_recent_mail(aid, datetime.now(UTC)) # activity within TIER_HOT_ACTIVITY_H + await leases.mark_live(get_redis(), aid) + c = _fast(SyncCoordinator("w1")) + c.tier_eval_interval = 0.1 + await c.start() + try: + await _until(lambda: aid in c.owned) + await _until(lambda: connections[aid] == 1) # recent mail => hot + assert c._runners[aid].tier == "hot" + finally: + await c.stop() + + +async def test_user_active_event_promotes_immediately(connections, make_accounts) -> None: # type: ignore[no-untyped-def] + aid0, aid1 = await make_accounts(2) # both under one user + user_id = await _user_of(aid0) + await leases.mark_live(get_redis(), aid0) + await leases.mark_live(get_redis(), aid1) + c = _fast(SyncCoordinator("w1")) # tier eval disabled: only the event can promote + await c.start() + try: + await _until(lambda: {aid0, aid1} <= c.owned) + assert connections[aid0] == 0 and connections[aid1] == 0 # warm to start + await get_event_bus().publish("user.active", {"user_id": str(user_id)}) + await _until(lambda: connections[aid0] == 1 and connections[aid1] == 1, timeout_s=3.0) + finally: + await c.stop() + + +async def test_hot_budget_demotes_least_active(connections, make_accounts) -> None: # type: ignore[no-untyped-def] + aid_old, aid_new = await make_accounts(2) # activity-hot, same user, NO session + now = datetime.now(UTC) + await _add_recent_mail(aid_old, now - timedelta(hours=1)) # least recently active + await _add_recent_mail(aid_new, now) + await leases.mark_live(get_redis(), aid_old) + await leases.mark_live(get_redis(), aid_new) + c = _fast(SyncCoordinator("w1")) + c.tier_eval_interval = 0.1 + c.hot_budget = 1 # only one may be hot + await c.start() + try: + await _until(lambda: {aid_old, aid_new} <= c.owned) + # The newer account stays hot; the older is shed to warm by the budget. + await _until(lambda: c._runners[aid_new].tier == "hot") + await _until(lambda: c._runners[aid_old].tier == "warm") + assert connections[aid_new] == 1 and connections[aid_old] == 0 + finally: + await c.stop() + + def test_worker_id_is_unique() -> None: assert coordinator_mod._new_worker_id() != coordinator_mod._new_worker_id()