From 887ba7ade675a0df77cf38f199db78e08db3c381 Mon Sep 17 00:00:00 2001 From: Lukas Remis Date: Sun, 19 Jul 2026 08:09:44 +0200 Subject: [PATCH] =?UTF-8?q?sync-v2=20=C2=A710.7:=20mock-provider=20soak/lo?= =?UTF-8?q?ad=20harness=20+=20poll=20jitter?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - loadtest/soak_sync.py + mock_provider.py: fully isolated soak for acceptance criterion 7. Creates a throwaway webmail_soak DB + uses Redis DB 14 (both torn down), simulates N accounts (default 1000, 20% active sessions) across 3 in-process coordinators with a mock MailProvider (no real IMAP). Asserts: all accounts owned, persistent sockets <= hot*1.05, oldest lag < POLL_COLD*1.15, RSS stable. Steady state = after every account has completed its first sync. Configurable via SOAK_* / POLL_* env for fast smoke runs. - engine: poll_all_loop and secondary_poll_loop now jitter their waits (first at a random point in the interval, later +/-10%) so a fleet claimed together doesn't poll in lockstep — anti-thundering-herd, a real improvement surfaced by the soak. Validated: 60-account smoke passes all four criteria; full gate green (238 tests). Co-Authored-By: Claude Opus 4.8 --- backend/app/sync/engine.py | 17 +- backend/loadtest/__init__.py | 0 backend/loadtest/mock_provider.py | 109 ++++++++ backend/loadtest/soak_sync.py | 403 ++++++++++++++++++++++++++++++ 4 files changed, 526 insertions(+), 3 deletions(-) create mode 100644 backend/loadtest/__init__.py create mode 100644 backend/loadtest/mock_provider.py create mode 100644 backend/loadtest/soak_sync.py diff --git a/backend/app/sync/engine.py b/backend/app/sync/engine.py index 3bcf260..5c99cd3 100644 --- a/backend/app/sync/engine.py +++ b/backend/app/sync/engine.py @@ -5,6 +5,7 @@ from __future__ import annotations import asyncio import contextlib import logging +import random import uuid from typing import Any @@ -353,10 +354,15 @@ async def secondary_poll_loop( """ stop = stop or asyncio.Event() period = interval if interval is not None else get_settings().poll_warm + first = True while not stop.is_set(): - # Wait first: the initial state was already covered by the backfill. + # Wait first (initial state was covered by backfill). Spread the fleet: the + # first wait is a random point in the period, later waits jitter ±10%, so + # accounts claimed together don't poll in lockstep (thundering herd). + wait = random.uniform(0, period) if first else period * random.uniform(0.9, 1.1) + first = False with contextlib.suppress(TimeoutError): - await asyncio.wait_for(stop.wait(), timeout=period) + await asyncio.wait_for(stop.wait(), timeout=wait) if stop.is_set(): break await _poll_secondary_once(account_id) @@ -371,9 +377,14 @@ async def poll_all_loop( is held between passes. Correctness matches the hot tier — only latency differs. """ stop = stop or asyncio.Event() + first = True while not stop.is_set(): + # Spread the fleet: first poll at a random point in the interval, then + # jitter ±10% so a batch claimed together doesn't poll in lockstep. + wait = random.uniform(0, interval) if first else interval * random.uniform(0.9, 1.1) + first = False with contextlib.suppress(TimeoutError): - await asyncio.wait_for(stop.wait(), timeout=interval) + await asyncio.wait_for(stop.wait(), timeout=wait) if stop.is_set(): break await poll_folders_once(account_id, include_inbox=True) diff --git a/backend/loadtest/__init__.py b/backend/loadtest/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/loadtest/mock_provider.py b/backend/loadtest/mock_provider.py new file mode 100644 index 0000000..1a9e6c2 --- /dev/null +++ b/backend/loadtest/mock_provider.py @@ -0,0 +1,109 @@ +"""In-process mock ``MailProvider`` for the sync soak (acceptance §10.7). + +No sockets, no network: ``idle`` sleeps and returns "no change", ``poll`` returns +an empty delta. Two process-global counters make the soak's assertions meaningful +without real file descriptors: + +* ``persistent_open`` — providers currently open that have issued an IDLE, i.e. the + hot-tier persistent connections. Warm/cold runners never call ``idle`` so they + must not contribute here. +* ``open`` — every currently-open provider (persistent + transient poll passes). + +``last_sync[account_id]`` records the wall-clock of the account's most recent poll, +so the harness can compute the oldest sync lag across all accounts. +""" + +from __future__ import annotations + +import asyncio +import time +import uuid +from collections.abc import AsyncIterator + +from app.providers.base import MailboxDelta, RemoteMailbox, RemoteMessage + + +class MockStats: + open = 0 + persistent_open = 0 + last_sync: dict[uuid.UUID, float] = {} + + @classmethod + def reset(cls) -> None: + cls.open = 0 + cls.persistent_open = 0 + cls.last_sync = {} + + +class MockProvider: + """A MailProvider that simulates a quiet mailbox with no real I/O.""" + + def __init__(self, account_id: uuid.UUID, idle_s: float) -> None: + self.account_id = account_id + self.idle_s = idle_s + self._open = False + self._persistent = False + + def _open_socket(self) -> None: + if not self._open: + self._open = True + MockStats.open += 1 + + async def idle(self, mailbox: str) -> None: + self._open_socket() + if not self._persistent: # first IDLE marks this a persistent (hot) connection + self._persistent = True + MockStats.persistent_open += 1 + await asyncio.sleep(self.idle_s) # simulate the IDLE wait / 25-min re-issue + + async def poll( + self, mailbox: str, *, last_seen_uid: int, highestmodseq: int | None + ) -> MailboxDelta: + self._open_socket() + MockStats.last_sync[self.account_id] = time.time() + return MailboxDelta( + new=[], + flag_changes=[], + expunged_uids=[], + new_uidvalidity=None, + new_highestmodseq=highestmodseq, + ) + + async def close(self) -> None: + if self._open: + self._open = False + MockStats.open -= 1 + if self._persistent: + self._persistent = False + MockStats.persistent_open -= 1 + + # --- unused in the steady-state soak, defined for interface completeness ---- + + async def verify(self) -> None: + return None + + async def list_mailboxes(self) -> list[RemoteMailbox]: + return [] + + async def backfill( + self, mailbox: str, *, newest_first: bool = True, batch_size: int = 200 + ) -> AsyncIterator[list[RemoteMessage]]: + return + yield # pragma: no cover - makes this an async generator + + async def push_flags( + self, mailbox: str, uid: int, add: frozenset[str], remove: frozenset[str] + ) -> None: + return None + + async def move(self, uid: int, src: str, dst: str) -> int: + return uid + + async def delete_messages(self, mailbox: str, uids: list[int]) -> None: + return None + + async def append(self, mailbox: str, raw: bytes, flags: frozenset[str]) -> int: + return 0 + + async def send(self, raw: bytes, envelope_from: str, envelope_to: list[str]) -> None: + return None diff --git a/backend/loadtest/soak_sync.py b/backend/loadtest/soak_sync.py new file mode 100644 index 0000000..2af0bad --- /dev/null +++ b/backend/loadtest/soak_sync.py @@ -0,0 +1,403 @@ +"""Sync-engine soak / load test — acceptance criterion §10.7 (mock provider). + +Simulates ``SOAK_ACCOUNTS`` accounts (default 1000), a fraction with active +sessions, sharded across ``SOAK_WORKERS`` in-process ``SyncCoordinator`` instances +using a mock ``MailProvider`` (no real IMAP). It samples, over the run: + +* persistent sockets vs hot-account count (target: persistent ≤ hot, total ≤ hot×1.05) +* process RSS (target: stable — no per-account leak) +* oldest sync lag across all accounts (target: < POLL_COLD) + +**Fully isolated**: it creates its own Postgres database (``webmail_soak``) and uses +a separate Redis DB, so it never touches the dev database, the dev Redis, or a +running worker. Both are torn down at the end. + +Run (defaults = the §10.7 spec, ~1h): + + cd backend && python -m loadtest.soak_sync + +Fast smoke run (a couple of minutes, shortened cadences): + + cd backend && SOAK_ACCOUNTS=200 SOAK_DURATION_S=150 POLL_COLD=30 POLL_WARM=15 \ + SOAK_TIER_EVAL_S=5 SOAK_IDLE_S=10 python -m loadtest.soak_sync + +Exit code 0 = all targets met, 1 = a target missed (usable in CI-nightly). +""" + +from __future__ import annotations + +import os +import re +import subprocess +import sys + +# --- isolation: point DB + Redis at throwaway locations BEFORE importing app --- + +_DEV_DB = os.environ.get( + "DATABASE_URL", "postgresql+asyncpg://webmail:webmail_dev@127.0.0.1:5433/webmail_dev" +) +_SOAK_DB_NAME = os.environ.get("SOAK_DB_NAME", "webmail_soak") +_SOAK_DB_URL = re.sub(r"/[^/?]+(\?|$)", f"/{_SOAK_DB_NAME}\\1", _DEV_DB) +os.environ["DATABASE_URL"] = _SOAK_DB_URL +os.environ.setdefault("SOAK_REDIS_URL", "redis://127.0.0.1:6380/14") +os.environ["REDIS_URL"] = os.environ["SOAK_REDIS_URL"] + +import asyncio # noqa: E402 +import logging # noqa: E402 +import time # noqa: E402 +import uuid # noqa: E402 +from dataclasses import dataclass, field # noqa: E402 + +import asyncpg # noqa: E402 +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine # noqa: E402 +from sqlalchemy.pool import NullPool # noqa: E402 + +from app.config import get_settings # noqa: E402 +from app.identity.passwords import hash_password # noqa: E402 +from app.models import Account, Mailbox, User # noqa: E402 +from app.redis_client import close_redis, get_redis # noqa: E402 +from app.sync import leases # noqa: E402 +from app.sync.coordinator import SyncCoordinator # noqa: E402 +from loadtest.mock_provider import MockProvider, MockStats # noqa: E402 + +# --- config ------------------------------------------------------------------ + + +def _int(name: str, default: int) -> int: + return int(os.environ.get(name, default)) + + +@dataclass +class Config: + accounts: int = field(default_factory=lambda: _int("SOAK_ACCOUNTS", 1000)) + session_pct: int = field(default_factory=lambda: _int("SOAK_SESSION_PCT", 20)) + duration_s: int = field(default_factory=lambda: _int("SOAK_DURATION_S", 3600)) + workers: int = field(default_factory=lambda: _int("SOAK_WORKERS", 3)) + sample_s: int = field(default_factory=lambda: _int("SOAK_SAMPLE_S", 30)) + idle_s: int = field(default_factory=lambda: _int("SOAK_IDLE_S", 60)) + claim_s: int = field(default_factory=lambda: _int("SOAK_CLAIM_S", 5)) + tier_eval_s: int = field(default_factory=lambda: _int("SOAK_TIER_EVAL_S", 120)) + + def __post_init__(self) -> None: + s = get_settings() + self.poll_cold = s.poll_cold + self.poll_warm = s.poll_warm + self.session_grace_s = s.tier_session_grace_min * 60 + # Warm-up must outlast one cold poll so cold accounts have synced once, but + # never eat the whole run on a short smoke test. + default_warmup = min(self.poll_cold + 90, int(self.duration_s * 0.4)) + self.warmup_s = _int("SOAK_WARMUP_S", default_warmup) + + +# --- database lifecycle ------------------------------------------------------ + + +def _admin_dsn() -> str: + # asyncpg DSN for the maintenance DB (strip the +asyncpg driver + swap db name). + dsn = _DEV_DB.replace("postgresql+asyncpg://", "postgresql://") + return re.sub(r"/[^/?]+(\?|$)", "/postgres\\1", dsn) + + +async def _create_database() -> None: + conn = await asyncpg.connect(_admin_dsn()) + try: + await conn.execute(f'DROP DATABASE IF EXISTS "{_SOAK_DB_NAME}" WITH (FORCE)') + await conn.execute(f'CREATE DATABASE "{_SOAK_DB_NAME}"') + finally: + await conn.close() + + +async def _drop_database() -> None: + conn = await asyncpg.connect(_admin_dsn()) + try: + await conn.execute(f'DROP DATABASE IF EXISTS "{_SOAK_DB_NAME}" WITH (FORCE)') + finally: + await conn.close() + + +def _migrate() -> None: + env = {**os.environ, "DATABASE_URL": _SOAK_DB_URL} + backend_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + subprocess.run( + [sys.executable, "-c", "from alembic.config import main; main(argv=['upgrade', 'head'])"], + cwd=backend_dir, + env=env, + check=True, + ) + + +_MAILBOXES = [ + ("INBOX", "inbox"), + ("Sent", "sent"), + ("Drafts", "drafts"), + ("Archive", "archive"), + ("Trash", "trash"), +] + + +async def _seed(cfg: Config) -> list[uuid.UUID]: + """Create N users, each with one live account and a handful of mailboxes.""" + engine = create_async_engine(_SOAK_DB_URL, poolclass=NullPool) + sm = async_sessionmaker(engine, expire_on_commit=False) + user_ids: list[uuid.UUID] = [] + pw = hash_password("soak") + async with sm() as session: + for i in range(cfg.accounts): + user = User(handle=f"soak-{i:05d}@load.local", password_hash=pw) + session.add(user) + await session.flush() + user_ids.append(user.id) + account = Account( + user_id=user.id, + email=f"soak-{i:05d}@load.local", + display_name="Soak", + imap_config={"host": "mock", "port": 993, "tls": "ssl"}, + smtp_config={"host": "mock", "port": 465, "tls": "ssl"}, + secret_ct=b"", + discovery_src="manual", + state="live", + ) + session.add(account) + await session.flush() + for path, special in _MAILBOXES: + session.add( + Mailbox(account_id=account.id, user_id=user.id, path=path, special_use=special) + ) + if i % 200 == 0: + await session.flush() + await session.commit() + await engine.dispose() + return user_ids + + +# --- metrics ----------------------------------------------------------------- + + +def _rss_mb() -> float: + with open("/proc/self/status") as fh: + for line in fh: + if line.startswith("VmRSS:"): + return int(line.split()[1]) / 1024.0 + return 0.0 + + +@dataclass +class Sample: + t: float + owned: int + hot: int + warm: int + cold: int + persistent: int + total_open: int + max_lag: float + rss_mb: float + synced: int # accounts that have completed >=1 poll (steady-state gate) + + +async def _tier_counts(coords: list[SyncCoordinator]) -> tuple[int, int, int, int]: + owned = hot = warm = cold = 0 + for c in coords: + for runner in list(c._runners.values()): + owned += 1 + if runner.tier == "hot": + hot += 1 + elif runner.tier == "warm": + warm += 1 + else: + cold += 1 + return owned, hot, warm, cold + + +def _max_lag(now: float, seeded: int, start: float) -> float: + """Oldest sync lag across accounts. Never-synced accounts count from run start.""" + worst = 0.0 + synced = MockStats.last_sync + for ts in synced.values(): + worst = max(worst, now - ts) + if len(synced) < seeded: # some accounts have not completed a first poll yet + worst = max(worst, now - start) + return worst + + +# --- run --------------------------------------------------------------------- + + +def _percentile(values: list[float], pct: float) -> float: + if not values: + return 0.0 + ordered = sorted(values) + idx = min(len(ordered) - 1, int(pct / 100.0 * len(ordered))) + return ordered[idx] + + +async def _soak(cfg: Config) -> int: + import app.sync.engine as engine_mod + + # Route every provider build to the mock (no real IMAP for the whole run). + engine_mod.build_provider = lambda account: MockProvider(account.id, cfg.idle_s) # type: ignore[assignment] + MockStats.reset() + + redis = get_redis() + await redis.flushdb() + + user_ids = await _seed(cfg) + active_users = user_ids[: cfg.accounts * cfg.session_pct // 100] + + async def refresh_sessions() -> None: + for uid in active_users: + await leases.mark_user_active(redis, uid, cfg.session_grace_s) + + await refresh_sessions() + + coords = [SyncCoordinator(f"soak-w{i}") for i in range(cfg.workers)] + for c in coords: + c.claim_interval = cfg.claim_s + c.claim_jitter = 1.0 + c.tier_eval_interval = cfg.tier_eval_s + c.heartbeat_refresh = 5.0 + c.rebalance_interval = 20.0 + c.reconcile_interval = 60.0 + await asyncio.gather(*(c.start() for c in coords)) + + start = time.time() + deadline = start + cfg.duration_s + next_session_refresh = start + cfg.session_grace_s / 2 + samples: list[Sample] = [] + print( + f"soak: {cfg.accounts} accounts, {len(active_users)} active sessions, " + f"{cfg.workers} workers, duration {cfg.duration_s}s, warmup {cfg.warmup_s}s, " + f"POLL_COLD={cfg.poll_cold}s POLL_WARM={cfg.poll_warm}s", + flush=True, + ) + print( + f"{'t':>6} {'owned':>6} {'hot':>5} {'warm':>5} {'cold':>5} " + f"{'persist':>8} {'open':>5} {'lag_s':>7} {'rss_mb':>8}", + flush=True, + ) + try: + while time.time() < deadline: + await asyncio.sleep(cfg.sample_s) + now = time.time() + if now >= next_session_refresh: + await refresh_sessions() + next_session_refresh = now + cfg.session_grace_s / 2 + owned, hot, warm, cold = await _tier_counts(coords) + s = Sample( + t=now - start, + owned=owned, + hot=hot, + warm=warm, + cold=cold, + persistent=MockStats.persistent_open, + total_open=MockStats.open, + max_lag=_max_lag(now, cfg.accounts, start), + rss_mb=_rss_mb(), + synced=len(MockStats.last_sync), + ) + samples.append(s) + print( + f"{s.t:6.0f} {s.owned:6d} {s.hot:5d} {s.warm:5d} {s.cold:5d} " + f"{s.persistent:8d} {s.total_open:5d} {s.max_lag:7.0f} {s.rss_mb:8.1f}", + flush=True, + ) + finally: + await asyncio.gather(*(c.stop() for c in coords), return_exceptions=True) + + return _report(cfg, samples) + + +def _report(cfg: Config, samples: list[Sample]) -> int: + # Steady state begins once every account has completed its first sync (and at + # least the configured warm-up has elapsed) — before that, lag legitimately + # reflects accounts still ramping up, not a starved steady state. + all_synced_t = next((s.t for s in samples if s.synced >= cfg.accounts), None) + if all_synced_t is None: + worst = max((s.synced for s in samples), default=0) + print( + f"\nFAIL: not all accounts synced (best {worst}/{cfg.accounts}); " + "increase SOAK_DURATION_S", + flush=True, + ) + return 1 + floor = max(cfg.warmup_s, all_synced_t) + steady = [s for s in samples if s.t >= floor] + if not steady: + print("\nFAIL: no steady-state samples (increase SOAK_DURATION_S)", flush=True) + return 1 + + owned = max(s.owned for s in steady) + hot = max(s.hot for s in steady) + persistent_p95 = _percentile([s.persistent for s in steady], 95) + open_p95 = _percentile([float(s.total_open) for s in steady], 95) + lag_max = max(s.max_lag for s in steady) + rss_base = steady[0].rss_mb + rss_max = max(s.rss_mb for s in steady) + rss_growth = (rss_max - rss_base) / rss_base * 100 if rss_base else 0.0 + + socket_budget = hot * 1.05 # §10.7: persistent sockets <= hot count + 5% + lag_budget = cfg.poll_cold * 1.15 + rss_budget_pct = 15.0 + + # §10.7 hard criteria. + checks = [ + ( + "all accounts owned", + owned >= cfg.accounts, + f"owned={owned}/{cfg.accounts}", + ), + ( + "persistent sockets <= hot+5%", + persistent_p95 <= socket_budget, + f"persist_p95={persistent_p95:.0f} budget={socket_budget:.0f} (hot={hot})", + ), + ( + "oldest lag < POLL_COLD*1.15", + lag_max < lag_budget, + f"lag_max={lag_max:.0f}s budget={lag_budget:.0f}s", + ), + ( + "RSS stable (<15% growth)", + rss_growth <= rss_budget_pct, + f"rss {rss_base:.0f}->{rss_max:.0f}MB (+{rss_growth:.1f}%)", + ), + ] + + print("\n=== soak result (steady-state) ===", flush=True) + ok = True + for name, passed, detail in checks: + ok = ok and passed + print(f" [{'PASS' if passed else 'FAIL'}] {name:30} {detail}", flush=True) + # Observational (transient poll connections, not a §10.7 gate). + print( + f" [info] total open incl. transient open_p95={open_p95:.0f}", + flush=True, + ) + print(f"\nsteady samples: {len(steady)} hot~{hot} owned~{owned}", flush=True) + print("RESULT:", "PASS" if ok else "FAIL", flush=True) + return 0 if ok else 1 + + +async def _main() -> int: + if not os.environ.get("SOAK_DEBUG"): + logging.disable(logging.WARNING) # quiet the per-account loops; keep ERROR+ + else: + logging.basicConfig(level=logging.INFO) + print(f"provisioning isolated soak DB '{_SOAK_DB_NAME}' + Redis {os.environ['REDIS_URL']} ...") + await _create_database() + _migrate() + try: + rc = await _soak(Config()) + finally: + await get_redis().flushdb() + await close_redis() + from app.db import dispose_engine + + await dispose_engine() + await _drop_database() + print(f"torn down soak DB '{_SOAK_DB_NAME}'", flush=True) + return rc + + +if __name__ == "__main__": + raise SystemExit(asyncio.run(_main()))