From 3d962a183c774eafa13170953eae1242dd4f44b2 Mon Sep 17 00:00:00 2001 From: Lukas Remis Date: Sun, 19 Jul 2026 09:28:48 +0200 Subject: [PATCH] =?UTF-8?q?sync-v2=20=C2=A710.7:=20mail-injecting=20mode?= =?UTF-8?q?=20for=20the=20soak=20harness?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Optional SOAK_MAIL=1 makes the mock deliver synthetic messages (unique UIDs, well-formed RFC822) on INBOX polls, so the full parse -> blob -> DB -> index path actually runs and CPU/RSS move under real throughput. Tunables: SOAK_MAIL_PER_POLL, SOAK_MSG_KB, SOAK_BIG_MSG_PCT (percent above STREAM_PARSE_MB -> streaming/§8 path). Blob store + search backend are isolated (temp BLOB_ROOT, SEARCH_BACKEND=pg) and torn down. Output gains msgs + msg/s columns and a delivered/peak-RSS summary. RSS check is now a trend (early-third vs late-third mean, not min-to-max) so it detects a real leak without flaking on transient large-message parse spikes. Validated: mail run delivers ~4.2 msg/s (incl. 5% 10MB messages); RSS oscillates 123-267MB but its mean is flat (+3.7% over the run) -> no leak, parse memory bound holds (§8); tier re-promotion on new mail exercised; all §10.7 checks pass. Full gate green (238 tests). Co-Authored-By: Claude Opus 4.8 --- backend/loadtest/mock_provider.py | 51 ++++++++++++++++++++- backend/loadtest/soak_sync.py | 74 +++++++++++++++++++++++++++---- 2 files changed, 115 insertions(+), 10 deletions(-) diff --git a/backend/loadtest/mock_provider.py b/backend/loadtest/mock_provider.py index 1a9e6c2..947431e 100644 --- a/backend/loadtest/mock_provider.py +++ b/backend/loadtest/mock_provider.py @@ -16,9 +16,13 @@ so the harness can compute the oldest sync lag across all accounts. from __future__ import annotations import asyncio +import random import time import uuid from collections.abc import AsyncIterator +from datetime import UTC, datetime +from email.message import EmailMessage +from email.utils import format_datetime from app.providers.base import MailboxDelta, RemoteMailbox, RemoteMessage @@ -28,11 +32,40 @@ class MockStats: persistent_open = 0 last_sync: dict[uuid.UUID, float] = {} + # --- optional mail injection (exercises parse/blob/DB/index under load) ---- + mail_enabled = False + mail_per_poll = 0.0 # expected new INBOX messages per poll (fractional => prob) + msg_bytes = 10_000 + big_pct = 0 # percent of messages that exceed STREAM_PARSE_MB (streaming path) + big_bytes = 10_000_000 + delivered = 0 # total synthetic messages handed to the sync engine + _uid: dict[uuid.UUID, int] = {} + @classmethod def reset(cls) -> None: cls.open = 0 cls.persistent_open = 0 cls.last_sync = {} + cls.delivered = 0 + cls._uid = {} + + +def _make_raw(size_bytes: int) -> bytes: + """A unique, well-formed RFC822 message padded to roughly ``size_bytes``.""" + msg = EmailMessage() + msg["Subject"] = f"soak {uuid.uuid4().hex[:12]}" + msg["From"] = "sender@load.local" + msg["To"] = "rcpt@load.local" + msg["Message-ID"] = f"<{uuid.uuid4().hex}@load.local>" + msg["Date"] = format_datetime(datetime.now(UTC)) + msg.set_content("x" * max(0, size_bytes - 256)) + return msg.as_bytes() + + +def _draw_count() -> int: + n = int(MockStats.mail_per_poll) + frac = MockStats.mail_per_poll - n + return n + (1 if random.random() < frac else 0) class MockProvider: @@ -61,8 +94,24 @@ class MockProvider: ) -> MailboxDelta: self._open_socket() MockStats.last_sync[self.account_id] = time.time() + new: list[RemoteMessage] = [] + if MockStats.mail_enabled and mailbox == "INBOX": + for _ in range(_draw_count()): + uid = MockStats._uid.get(self.account_id, 0) + 1 + MockStats._uid[self.account_id] = uid + big = random.random() < MockStats.big_pct / 100.0 + size = MockStats.big_bytes if big else MockStats.msg_bytes + new.append( + RemoteMessage( + uid=uid, + raw=_make_raw(size), + flags=frozenset(), + internal_date=datetime.now(UTC), + ) + ) + MockStats.delivered += len(new) return MailboxDelta( - new=[], + new=new, flag_changes=[], expunged_uids=[], new_uidvalidity=None, diff --git a/backend/loadtest/soak_sync.py b/backend/loadtest/soak_sync.py index 80f73da..02c10c4 100644 --- a/backend/loadtest/soak_sync.py +++ b/backend/loadtest/soak_sync.py @@ -21,6 +21,15 @@ 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 +Mail-injecting mode (SOAK_MAIL=1) delivers synthetic messages so the parse / +blob / DB / index path actually runs and RSS + CPU move under throughput. Tunables: +SOAK_MAIL_PER_POLL (expected new INBOX msgs per poll), SOAK_MSG_KB, SOAK_BIG_MSG_PCT +(percent above STREAM_PARSE_MB, exercising the streaming/§8 path). Example: + + cd backend && SOAK_MAIL=1 SOAK_MAIL_PER_POLL=0.5 SOAK_BIG_MSG_PCT=5 \ + SOAK_ACCOUNTS=300 SOAK_DURATION_S=300 POLL_COLD=60 POLL_WARM=30 \ + SOAK_TIER_EVAL_S=15 SOAK_IDLE_S=20 python -m loadtest.soak_sync + Exit code 0 = all targets met, 1 = a target missed (usable in CI-nightly). """ @@ -28,10 +37,12 @@ from __future__ import annotations import os import re +import shutil import subprocess import sys +import tempfile -# --- isolation: point DB + Redis at throwaway locations BEFORE importing app --- +# --- isolation: point DB + Redis + blobs at throwaway locations BEFORE app import _DEV_DB = os.environ.get( "DATABASE_URL", "postgresql+asyncpg://webmail:webmail_dev@127.0.0.1:5433/webmail_dev" @@ -41,6 +52,11 @@ _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"] +# Isolate the blob store (mail injection writes real blobs) and force the in-DB +# search backend so nothing leaks to the dev blob dir or an external Meilisearch. +_SOAK_BLOBS = tempfile.mkdtemp(prefix="soak-blobs-") +os.environ["BLOB_ROOT"] = _SOAK_BLOBS +os.environ["SEARCH_BACKEND"] = "pg" import asyncio # noqa: E402 import logging # noqa: E402 @@ -77,6 +93,14 @@ class Config: 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)) + # Mail injection (off by default): synthetic messages exercise the + # parse/blob/DB/index path so RSS + CPU move under real throughput. + mail: bool = field(default_factory=lambda: bool(_int("SOAK_MAIL", 0))) + mail_per_poll: float = field( + default_factory=lambda: float(os.environ.get("SOAK_MAIL_PER_POLL", 0.3)) + ) + msg_kb: int = field(default_factory=lambda: _int("SOAK_MSG_KB", 12)) + big_pct: int = field(default_factory=lambda: _int("SOAK_BIG_MSG_PCT", 0)) def __post_init__(self) -> None: s = get_settings() @@ -194,6 +218,7 @@ class Sample: rss_mb: float synced: int # accounts that have completed >=1 poll (steady-state gate) dead: int # owned runners whose work loops have died (zombies) + delivered: int # cumulative synthetic messages persisted (mail mode) async def _tier_counts(coords: list[SyncCoordinator]) -> tuple[int, int, int, int, int]: @@ -240,6 +265,12 @@ async def _soak(cfg: Config) -> int: # 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() + if cfg.mail: + MockStats.mail_enabled = True + MockStats.mail_per_poll = cfg.mail_per_poll + MockStats.msg_bytes = cfg.msg_kb * 1024 + MockStats.big_pct = cfg.big_pct + MockStats.big_bytes = (get_settings().stream_parse_mb + 2) * 1024 * 1024 redis = get_redis() await redis.flushdb() @@ -274,15 +305,21 @@ async def _soak(cfg: Config) -> int: deadline = start + cfg.duration_s next_session_refresh = start + cfg.session_grace_s / 2 samples: list[Sample] = [] + mail_desc = ( + f"mail ON ~{cfg.mail_per_poll}/poll {cfg.msg_kb}KB big={cfg.big_pct}%" + if cfg.mail + else "mail OFF" + ) 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", + f"POLL_COLD={cfg.poll_cold}s POLL_WARM={cfg.poll_warm}s, {mail_desc}", 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} {'dead':>5}", + f"{'persist':>8} {'open':>5} {'lag_s':>7} {'rss_mb':>8} {'dead':>5} " + f"{'msgs':>8} {'msg/s':>7}", flush=True, ) try: @@ -305,12 +342,15 @@ async def _soak(cfg: Config) -> int: rss_mb=_rss_mb(), synced=len(MockStats.last_sync), dead=dead, + delivered=MockStats.delivered, ) samples.append(s) + prev = samples[-2] if len(samples) > 1 else None + rate = (s.delivered - prev.delivered) / cfg.sample_s if prev else 0.0 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} " - f"{s.dead:5d}", + f"{s.dead:5d} {s.delivered:8d} {rate:7.1f}", flush=True, ) finally: @@ -343,9 +383,14 @@ def _report(cfg: Config, samples: list[Sample]) -> int: 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 trend: compare the mean of the first third of steady samples to the last + # third. This detects a genuine upward drift (leak) while ignoring transient + # parse spikes from large messages (min-to-max would flake in mail mode). + third = max(1, len(steady) // 3) + rss_early = sum(s.rss_mb for s in steady[:third]) / third + rss_late = sum(s.rss_mb for s in steady[-third:]) / third 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 + rss_growth = (rss_late - rss_early) / rss_early * 100 if rss_early else 0.0 socket_budget = hot * 1.05 # §10.7: persistent sockets <= hot count + 5% lag_budget = cfg.poll_cold * 1.15 @@ -369,9 +414,10 @@ def _report(cfg: Config, samples: list[Sample]) -> int: f"lag_max={lag_max:.0f}s budget={lag_budget:.0f}s", ), ( - "RSS stable (<15% growth)", + "RSS trend stable (<15%)", rss_growth <= rss_budget_pct, - f"rss {rss_base:.0f}->{rss_max:.0f}MB (+{rss_growth:.1f}%)", + f"rss early {rss_early:.0f}MB -> late {rss_late:.0f}MB " + f"(+{rss_growth:.1f}%), peak {rss_max:.0f}MB", ), ] @@ -385,6 +431,15 @@ def _report(cfg: Config, samples: list[Sample]) -> int: f" [info] total open incl. transient open_p95={open_p95:.0f}", flush=True, ) + if cfg.mail: + delivered = samples[-1].delivered + span = samples[-1].t - samples[0].t or 1.0 + rss_peak = max(s.rss_mb for s in samples) + print( + f" [info] mail delivered={delivered} (~{delivered / span:.1f} msg/s), " + f"peak RSS={rss_peak:.0f}MB", + 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 @@ -407,7 +462,8 @@ async def _main() -> int: await dispose_engine() await _drop_database() - print(f"torn down soak DB '{_SOAK_DB_NAME}'", flush=True) + shutil.rmtree(_SOAK_BLOBS, ignore_errors=True) + print(f"torn down soak DB '{_SOAK_DB_NAME}' + blob dir", flush=True) return rc