From 8eb699a05f20282017e972752fd3d02a08e006f1 Mon Sep 17 00:00:00 2001 From: Lukas Remis Date: Thu, 17 Sep 2026 21:04:38 +0200 Subject: [PATCH] Antispam: bootstrap every account, not just the ones we happen to see go live MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Warming Bayes hung on a single event — `account.state_changed` reaching "live". That covers an account added from now on and nothing else. An account that went live before this plugin existed never gets one; neither does one whose Junk folder was empty at the time and filled up later, or one whose owner enabled spam-folder sync afterwards. A worker restarting at the wrong moment loses it too. The result was a feature that quietly required someone to run it by hand per account, which is not a feature. An hourly sweep now walks every live account and offers each a bootstrap. That works because the per-account job was already cheap to repeat: it returns at once for an account already done, and declines — without marking it done — when the corpus is still too small to learn from. So the sweep simply keeps asking until there is something worth learning, then does it once. `s.accounts.live()` is the new SDK surface it needs: cross-user by design, because a sweep runs for the instance rather than a request. Ids and state only, so reading anything still goes through the user-scoped accessors. One failing account cannot stop the sweep — the mailbox that raises is logged and the rest still get their turn. docs/plugins.md now says this in general terms next to `JobSpec`: a state-change hook is one-shot, so pair it with a sweep when the work must eventually happen for everything. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_018SqJYt7jD7oUEkoY68sBZo --- backend/app/plugins/actions.py | 20 ++++++++ backend/app/plugins/sdk.py | 12 ++++- .../tests/plugins/test_antispam_learning.py | 48 +++++++++++++++++++ docs/plugins.md | 7 +++ .../webmail_plugin_antispam/__init__.py | 15 ++++-- .../webmail_plugin_antispam/jobs.py | 30 ++++++++++++ 6 files changed, 127 insertions(+), 5 deletions(-) diff --git a/backend/app/plugins/actions.py b/backend/app/plugins/actions.py index 16b966e..e937f5e 100644 --- a/backend/app/plugins/actions.py +++ b/backend/app/plugins/actions.py @@ -162,6 +162,26 @@ class PluginMessages: } +class PluginAccounts: + """Cross-user account queries, for scheduled jobs. + + Deliberately not user-scoped: a sweep runs on behalf of the instance, not a + request, and needs to find work across every user. Nothing here returns + content — only ids and state — so a plugin still has to go through the + user-scoped accessors to read anything. + """ + + async def live(self) -> list[tuple[uuid.UUID, uuid.UUID]]: + """``(user_id, account_id)`` for every account finished with its initial + import. Accounts still backfilling are excluded: acting on one means + acting on a partial mailbox.""" + async with session_scope() as session: + rows = await session.execute( + select(Account.user_id, Account.id).where(Account.state == "live") + ) + return [(u, a) for u, a in rows] + + class PluginThreads: """Read-only, user-scoped thread queries (e.g. self-healing stale references).""" diff --git a/backend/app/plugins/sdk.py b/backend/app/plugins/sdk.py index 99297e7..b74045c 100644 --- a/backend/app/plugins/sdk.py +++ b/backend/app/plugins/sdk.py @@ -28,7 +28,12 @@ from app.identity.deps import current_user, require_csrf from app.indexer import blobstore from app.models import User from app.outbox import service as outbox_service -from app.plugins.actions import PluginActions, PluginMessages, PluginThreads +from app.plugins.actions import ( + PluginAccounts, + PluginActions, + PluginMessages, + PluginThreads, +) from app.plugins.db import plugin_session from app.plugins.http import PluginHttp from app.plugins.kv import PluginKvStore @@ -153,6 +158,11 @@ class PluginSdk: ``move_to_special`` (one message into e.g. Junk).""" return PluginActions() + @property + def accounts(self) -> PluginAccounts: + """Cross-user account listing for scheduled jobs (`live()`).""" + return PluginAccounts() + @property def threads(self) -> PluginThreads: """Read-only thread queries (e.g. pruning stale references).""" diff --git a/backend/tests/plugins/test_antispam_learning.py b/backend/tests/plugins/test_antispam_learning.py index bfb7015..c14fde4 100644 --- a/backend/tests/plugins/test_antispam_learning.py +++ b/backend/tests/plugins/test_antispam_learning.py @@ -280,3 +280,51 @@ async def test_bootstrap_caps_ham_at_a_multiple_of_spam(monkeypatch: pytest.Monk assert taught.count(True) == 20 assert taught.count(False) == 60, "ham should be capped at 3x the spam count" assert marked, "a completed bootstrap must mark the account done" + + +# --- the sweep: bootstrapping must not depend on catching an event ----------- + + +async def test_sweep_visits_every_live_account(monkeypatch: pytest.MonkeyPatch) -> None: + """An account that went live before the plugin existed still gets a warm start.""" + visited: list[tuple[str, str]] = [] + accounts = [(uuid.uuid4(), uuid.uuid4()) for _ in range(3)] + + class _Accounts: + async def live(self) -> list[tuple[uuid.UUID, uuid.UUID]]: + return accounts + + class _Fake: + accounts = _Accounts() + + async def _one(_ctx: object, *, user_id: str, account_id: str) -> None: + visited.append((user_id, account_id)) + + monkeypatch.setattr(jobs, "_SDK", _Fake()) + monkeypatch.setattr(jobs, "bootstrap_account", _one) + await jobs.bootstrap_sweep({}) + assert visited == [(str(u), str(a)) for u, a in accounts] + + +async def test_one_failing_account_does_not_stop_the_sweep( + monkeypatch: pytest.MonkeyPatch, +) -> None: + done: list[str] = [] + accounts = [(uuid.uuid4(), uuid.uuid4()) for _ in range(3)] + + class _Accounts: + async def live(self) -> list[tuple[uuid.UUID, uuid.UUID]]: + return accounts + + class _Fake: + accounts = _Accounts() + + async def _one(_ctx: object, *, user_id: str, account_id: str) -> None: + if account_id == str(accounts[1][1]): + raise RuntimeError("that mailbox is on fire") + done.append(account_id) + + monkeypatch.setattr(jobs, "_SDK", _Fake()) + monkeypatch.setattr(jobs, "bootstrap_account", _one) + await jobs.bootstrap_sweep({}) + assert done == [str(accounts[0][1]), str(accounts[2][1])] diff --git a/docs/plugins.md b/docs/plugins.md index c48e5fb..4bdff1b 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -100,6 +100,7 @@ s = sdk("myplugin") | `await s.http_get(url)` | HTTPS GET through the SSRF guard (blocks private IPs) | | `s.messages` | read a message: `envelope`, `header`, `raw`, `sample` | | `s.actions` | act on a message: `set_flags`, `trash`, `move_to_special` | +| `s.accounts` | `live()` → `(user_id, account_id)` for every account past its initial import — cross-user, for scheduled jobs | Plus, for routers: `CurrentUserId` (auth dependency → the user's uuid) and `require_csrf` (guard mutating routes). @@ -243,6 +244,12 @@ jobs = [ enqueue-on-demand via `await sdk("myplugin").enqueue(sync_now, user_id=...)`. - `singleton=True`: at most one running instance per `(plugin, user_id)` (Redis lock). +A hook on a state change is a **one-shot** event: it cannot help anything that +already happened before your plugin was installed, and a worker restart can miss +it. If a job must eventually run for every account, pair the hook with a +scheduled sweep over `s.accounts.live()` and make the per-account work cheap to +repeat — the antispam bootstrap does exactly this. + Jobs share the worker pool + retry semantics and log under `plg.myplugin.*`. --- diff --git a/plugins/webmail_plugin_antispam/webmail_plugin_antispam/__init__.py b/plugins/webmail_plugin_antispam/webmail_plugin_antispam/__init__.py index 2242d23..9791d44 100644 --- a/plugins/webmail_plugin_antispam/webmail_plugin_antispam/__init__.py +++ b/plugins/webmail_plugin_antispam/webmail_plugin_antispam/__init__.py @@ -25,7 +25,7 @@ from webmail_plugin_antispam.hook import ( on_message_changed, on_message_new, ) -from webmail_plugin_antispam.jobs import bootstrap_account +from webmail_plugin_antispam.jobs import bootstrap_account, bootstrap_sweep from webmail_plugin_antispam.router import router _HERE = Path(__file__).parent @@ -41,9 +41,16 @@ plugin = Plugin( # An account going live is the cue to warm Bayes from its history (§C.5). "account.state_changed": on_account_state_changed, }, - # Enqueue-on-demand (no schedule); singleton so a flapping account cannot - # start two training passes over the same mailbox. - jobs=[JobSpec(func=bootstrap_account, schedule=None, singleton=True)], + jobs=[ + # Enqueue-on-demand; singleton so a flapping account cannot start two + # training passes over the same mailbox. + JobSpec(func=bootstrap_account, schedule=None, singleton=True), + # And hourly, because the state-change hook is a one-shot event: it + # cannot help an account that went live before this plugin existed, or + # one whose Junk folder only became worth reading later. Offset from the + # hour so it does not pile onto every other scheduled job. + JobSpec(func=bootstrap_sweep, schedule="17 * * * *", singleton=True), + ], router=router, migrations=str(_HERE / "migrations" / "versions"), ui_manifest=UiManifest( diff --git a/plugins/webmail_plugin_antispam/webmail_plugin_antispam/jobs.py b/plugins/webmail_plugin_antispam/webmail_plugin_antispam/jobs.py index 289ba47..8468ed8 100644 --- a/plugins/webmail_plugin_antispam/webmail_plugin_antispam/jobs.py +++ b/plugins/webmail_plugin_antispam/webmail_plugin_antispam/jobs.py @@ -134,3 +134,33 @@ async def bootstrap_account( "ham": ham_taught, }, ) + + +async def bootstrap_sweep(ctx: dict[str, Any]) -> None: + """Find accounts that still need a warm start, and give them one. + + The `account.state_changed` hook catches an account the moment it finishes + importing, which covers accounts added from now on. This covers everything + else, and it is the reason bootstrapping is not a thing anyone has to + remember to do: + + * accounts that existed before this plugin did — their transition to `live` + happened long ago and will not happen again; + * accounts whose Junk folder was empty at the time and has since filled, or + which only started syncing the provider's spam folder later; + * a transition missed because the worker was restarting. + + Cheap to repeat: `bootstrap_account` returns immediately for an account + already marked done, and declines (without marking) one whose corpus is still + too small to learn from — so this simply retries until there is something + worth learning. + """ + for user_id, account_id in await _SDK.accounts.live(): + try: + await bootstrap_account( + ctx, user_id=str(user_id), account_id=str(account_id) + ) + except Exception as exc: # noqa: BLE001 - one bad account must not stop the sweep + logger.warning( + "antispam bootstrap sweep failed for %s: %s", account_id, exc + )