perf(users): password hashing off the event loop; cached admin user summary

CHANGELOG.md

@@ -81,6 +81,21 @@ All notable changes to this project are documented in this file. The format is b
 
 ### Fixed
 
+- Users password creation, forgot-password fingerprints, reset verification and
+  password updates offload hashing to worker threads, keeping unrelated async
+  requests responsive during registration and bulk invites. Password policy,
+  single-use reset tokens and atomic credential revocation are preserved.
+- Admin user state totals and role options use a 30-second per-worker cache.
+  A committed user, role or membership write evicts this worker's entry; the
+  admin service's write paths (create, invite, delete, disable, enable, mark
+  verified, set roles) also publish `users.admin_summary` once after commit so
+  other workers evict theirs. Other writes (logins, lockout counters, password
+  changes) publish nothing and rely on the TTL.
+  List DTOs skip redundant validation of trusted database values; role options
+  select only the required columns, and managers share one password helper.
+- Users: `on_commit` callbacks queued before a fastapi-users adapter commit
+  (user create, update, delete through the manager) now run after that commit;
+  they used to be discarded when the request session was finalized.
 - Include module `ui/` sources in wheel-installed Tailwind scanning and the asset manifest; retain legacy `components/` support.
 - Background tasks: the executions list (when empty) and the Workers page no
   longer wait out the broker's ~1 s inspect timeout before rendering. The

docs/modules/users.md

@@ -243,6 +243,28 @@ Everything else is DB-backed (initial values are pydantic defaults; edit under U
 | `Profile` | `/users/me` | `user` | `USER_DROPDOWN` | — | `990` | _logged-in_ |
 | `Logout` | `/users/logout` (POST) | `log-out` | `USER_DROPDOWN` | — | `999` | _logged-in_ |
 
+## Performance
+
+Password hashing and verification for local login, account creation (including
+registration and bulk invitations), password changes and reset fingerprints run
+in worker threads. Database operations, password policy and lifecycle hooks stay
+on the event loop; password resets still revoke credentials atomically.
+
+Admin user state totals and role options (including membership counts) are
+cached per worker for 30 seconds. A committed transaction that wrote a user,
+role or membership through the ORM evicts this worker's entry. Other workers
+hear about a change only through an explicit `users.admin_summary` publish from
+the admin service's write paths (create, invite, delete, disable, enable, mark
+verified, set roles), sent once after the commit and never on rollback. Other
+writes (logins, failed-login counters, password changes, session revocations,
+role edits made outside the users admin service) publish nothing, so another
+worker may show old totals until the 30-second TTL expires; the same holds
+without the Redis invalidation transport or when a message is lost.
+These summaries do not control permissions. Filtered user rows and pagination
+counts remain live queries. The accepted contains-search scan is unchanged.
+See [Users performance measurements](../perf/2026-11-05-users.md) for the local
+before/after results and reproduction command.
+
 ## Password reset and credential revocation
 
 A successful password reset, including one reached through an admin-generated

docs/perf/2026-11-05-users.md

@@ -0,0 +1,45 @@
+# Users performance measurements
+
+Measured locally on 2026-11-05, using production password costs
+(`SM_TEST_REAL_PASSWORD_HASHING=1`) and the installed fastapi-users 15.0.5.
+The baseline is the working tree before the password offload change.
+
+A concurrent heartbeat sleeps for 5 ms and records the excess scheduling delay
+while the real manager/service performs the operation. The bulk case calls
+`UserService.invite` sequentially for 100 distinct accounts; it does not include
+HTTP parsing or email delivery. Fixtures use SQLite, so these measurements do
+not reproduce the review's Postgres/10k-user workload.
+
+| Operation | Before: worst loop lag | After: worst loop lag | Before / after elapsed |
+|---|---:|---:|---:|
+| 100 sequential invitations | 254.1 ms | 5.1 ms | 27.273 / 27.387 s |
+| Forgot-password fingerprint | 237.2 ms | 4.4 ms | 0.267 / 0.256 s |
+
+Offloading keeps the event loop responsive; it does not reduce the underlying
+password hashing work. These are single-run observations, not request latency
+percentiles. This sandbox did not reliably wake its event loop after cross-thread
+completions (even a standalone aiosqlite connection hung). Both runs used the
+same temporary test-runner shim that capped selector waits at 1 ms; application
+code and hashing costs were unchanged by the shim.
+
+Run the opt-in measurements on a normal local environment:
+
+```sh
+SM_TEST_REAL_PASSWORD_HASHING=1 uv run pytest \
+  modules/users/tests/test_users_password_offload.py \
+  -k real_hashing -q -s -p no:cacheprovider
+```
+
+The ordinary targeted regressions check crypto thread identity independently
+of timing, reset-token reuse rejection, and that database execution stays on
+the event-loop thread. Existing credential-revocation and password-policy tests
+cover the reset's transaction semantics.
+
+For admin summaries, regressions record SQL execution and require **zero new
+queries** on repeated state-count and role-option reads within the TTL. They
+also cover expiry, local eviction when a user/role/membership write commits,
+one explicit `users.admin_summary` publish per admin write method after commit
+(none on rollback, none for login or lockout writes), remote invalidation,
+invalidation racing a cache fill, and role projections that do not instantiate
+ORM rows. DTO construction is checked without invoking SQLModel input
+validation.

modules/users/tests/test_users_admin_summary_cache.py

@@ -0,0 +1,264 @@
+"""Admin summaries avoid repeated scans, forget committed writes and publish explicitly."""
+
+import uuid
+from datetime import UTC, datetime
+from unittest.mock import patch
+
+import pytest
+from fastapi_users.password import PasswordHelper
+from simple_module_core.invalidation import Invalidation
+from simple_module_db.callbacks import register_on_commit
+from simple_module_db.transaction import finalize_session, rollback_session
+from sqlalchemy import event, select
+from users.admin import summary_cache
+from users.admin.service import UserService
+from users.contracts.schemas import UserCreate, UserListItem
+from users.db_adapter import UserDatabaseWithRoles
+from users.manager import UserManager
+from users.models import Role, User, UserRole
+from users.roles_cache import refresh_roles_cache
+
+
+def service(db, app):
+    manager = UserManager(
+        UserDatabaseWithRoles(db, User), app.state.users.mailer, app.state.users.settings
+    )
+    return UserService(db, manager, app)
+
+
+def _record(app):
+    messages = []
+    app.state.sm.invalidation.subscribe(summary_cache.CHANNEL, messages.append)
+    return messages
+
+
+async def _make_user(app, email="target@example.com", **fields) -> uuid.UUID:
+    async with app.state.sm.db.session_factory() as db:
+        user = User(email=email, hashed_password=PasswordHelper().hash("SecurePass1!"), **fields)
+        db.add(user)
+        await finalize_session(db)
+        return user.id
+
+
+@pytest.mark.parametrize("method", ["count_user_states", "list_roles"])
+async def test_summary_cache_hit_and_expiry(users_app, monkeypatch, method):
+    queries = []
+    event.listen(
+        users_app.state.sm.db.engine.sync_engine,
+        "before_cursor_execute",
+        lambda *args: queries.append(args[2]),
+    )
+    async with users_app.state.sm.db.session_factory() as db:
+        get = getattr(service(db, users_app), method)
+        first = await get()
+        queries.clear()
+        assert await get() == first
+        assert queries == []
+        # Values handed to callers must not let them corrupt the cached result.
+        first.clear()
+        assert await get()
+        monkeypatch.setattr(summary_cache.time, "monotonic", lambda: 10**12)
+        await get()
+        assert queries
+
+
+@pytest.mark.parametrize(
+    "write", ["create", "verify", "invite", "disable", "delete", "role", "membership"]
+)
+async def test_local_orm_writes_clear_this_workers_cache_on_commit(users_app, write):
+    async with users_app.state.sm.db.session_factory() as db:
+        svc = service(db, users_app)
+        user = User(
+            email="cached@example.com", hashed_password=PasswordHelper().hash("SecurePass1!")
+        )
+        db.add(user)
+        await finalize_session(db)
+        before = await svc.count_user_states()
+        roles_before = await svc.list_roles()
+        if write == "create":
+            db.add(User(email="added@example.com", hashed_password="unused"))
+        elif write == "verify":
+            user.is_verified = True
+        elif write == "invite":
+            user.invited_at = datetime.now(UTC)
+        elif write == "disable":
+            user.is_active = False
+        elif write == "delete":
+            await db.delete(user)
+        elif write == "role":
+            db.add(Role(name="reviewer"))
+        else:
+            role = (await db.execute(select(Role).where(Role.name == "user"))).scalar_one()
+            db.add(UserRole(user_id=user.id, role_id=role.id))
+        await finalize_session(db)
+        if write in ("role", "membership"):
+            assert await svc.list_roles() != roles_before
+        else:
+            assert await svc.count_user_states() != before
+
+
+async def test_rollback_and_remote_invalidation(users_app):
+    async with users_app.state.sm.db.session_factory() as db:
+        svc = service(db, users_app)
+        before = await svc.count_user_states()
+        db.add(User(email="rolledback@example.com", hashed_password="unused"))
+        await db.flush()
+        await rollback_session(db)
+        # A rollback must preserve a warm cache.
+        with patch.object(summary_cache, "load_counts", side_effect=AssertionError("cache miss")):
+            assert await svc.count_user_states() == before
+        await users_app.state.sm.invalidation.deliver(
+            Invalidation(summary_cache.CHANNEL, origin="other-worker").to_wire()
+        )
+        # Invalidate remotely even when no local write occurred.
+        with (
+            patch.object(summary_cache, "load_counts", side_effect=RuntimeError("reloaded")),
+            pytest.raises(RuntimeError, match="reloaded"),
+        ):
+            await svc.count_user_states()
+
+
+async def _call(svc, method, user_id):
+    if method == "create_user":
+        await svc.create_user("new@example.com", "SecurePass1!", None, ["user"], created_by=None)
+    elif method == "invite":
+        await svc.invite("invited@example.com", None, ["user"])
+    elif method == "set_roles":
+        await svc.set_roles(user_id, [])
+    else:
+        await getattr(svc, method)(user_id)
+
+
+_WRITES = ["create_user", "invite", "delete_user", "disable", "enable", "mark_verified"]
+
+
+@pytest.mark.parametrize("method", [*_WRITES, "set_roles"])
+async def test_admin_writes_publish_once_after_commit(users_app, method):
+    user_id = await _make_user(users_app, is_active=method != "enable")
+    messages = _record(users_app)
+    async with users_app.state.sm.db.session_factory() as db:
+        await _call(service(db, users_app), method, user_id)
+        assert messages == []
+        await finalize_session(db)
+    assert len(messages) == 1
+
+
+@pytest.mark.parametrize("method", [*_WRITES, "set_roles"])
+async def test_admin_writes_publish_nothing_on_rollback(users_app, method):
+    user_id = await _make_user(users_app, is_active=method != "enable")
+    messages = _record(users_app)
+    async with users_app.state.sm.db.session_factory() as db:
+        await _call(service(db, users_app), method, user_id)
+        await rollback_session(db)
+    assert messages == []
+
+
+async def test_login_and_lockout_writes_publish_nothing(users_app, anon_client):
+    messages = _record(users_app)
+    for _ in range(3):
+        resp = await anon_client.post(
+            "/api/users/auth/login",
+            data={"username": "admin@example.com", "password": "wrong-password"},
+        )
+        assert resp.status_code == 400, resp.text
+    resp = await anon_client.post(
+        "/api/users/auth/login",
+        data={"username": "admin@example.com", "password": "AdminPass1!"},
+    )
+    assert resp.status_code == 204, resp.text
+    assert messages == []
+
+
+@pytest.mark.parametrize("operation", ["create", "update", "delete"])
+async def test_adapter_commits_run_queued_on_commit_callbacks(users_app, operation):
+    ran = []
+    async with users_app.state.sm.db.session_factory() as db:
+        manager = service(db, users_app)._manager
+        register_on_commit(db, lambda: ran.append(operation))
+        if operation == "create":
+            await manager.create(
+                UserCreate(email="adapter@example.com", password="SecurePass1!", is_verified=True)
+            )
+        else:
+            user = await manager.get_by_email("admin@example.com")
+            if operation == "update":
+                await manager._update(user, {"is_active": False})
+            else:
+                await manager.delete(user)
+        assert ran == [operation]
+
+
+async def test_list_items_use_trusted_construction(users_app, monkeypatch):
+    def reject_validation(*args, **kwargs):
+        raise AssertionError("DB rows should not be validated again")
+
+    monkeypatch.setattr(UserListItem, "__init__", reject_validation)
+    async with users_app.state.sm.db.session_factory() as db:
+        svc = service(db, users_app)
+        items, total = await svc.list_users()
+        assert total and items[0].model_dump(mode="json")["email"]
+        user = await svc.get_with_roles(items[0].id)
+        assert svc.to_list_item(user) == items[0]
+
+
+async def test_inflight_invalidation_does_not_repopulate_stale_cache(users_app, monkeypatch):
+    calls = 0
+
+    async def racing_load(db):
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            await users_app.state.sm.invalidation.publish(summary_cache.CHANNEL)
+        return {"active": calls, "invited": 0, "unverified": 0}
+
+    monkeypatch.setattr(summary_cache, "load_counts", racing_load)
+    async with users_app.state.sm.db.session_factory() as db:
+        svc = service(db, users_app)
+        assert (await svc.count_user_states())["active"] == 1
+        assert (await svc.count_user_states())["active"] == 2
+        assert (await svc.count_user_states())["active"] == 2
+
+
+async def test_role_options_do_not_hydrate_orm_rows(users_app):
+    loaded = []
+
+    def record(*args):
+        loaded.append(args[0])
+
+    event.listen(Role, "load", record)
+    try:
+        async with users_app.state.sm.db.session_factory() as db:
+            assert await service(db, users_app).list_roles()
+            assert await refresh_roles_cache(users_app)
+        assert loaded == []
+    finally:
+        event.remove(Role, "load", record)
+
+
+async def test_create_user_without_roles_still_publishes(users_app):
+    messages = _record(users_app)
+    async with users_app.state.sm.db.session_factory() as db:
+        await service(db, users_app).create_user(
+            "noroles@example.com", "SecurePass1!", None, [], created_by=None
+        )
+        # The adapter committed; the request would roll back and drop callbacks.
+        await rollback_session(db)
+    assert len(messages) == 1
+
+
+async def test_login_bookkeeping_keeps_cache_warm_but_counted_columns_clear_it(users_app):
+    async with users_app.state.sm.db.session_factory() as db:
+        svc = service(db, users_app)
+        user = User(email="login@example.com", hashed_password="unused")
+        db.add(user)
+        await finalize_session(db)
+        before = await svc.count_user_states()
+        user.last_login_at = datetime.now(UTC)
+        user.failed_login_count = 3
+        user.session_version = (user.session_version or 0) + 1
+        await finalize_session(db)
+        with patch.object(summary_cache, "load_counts", side_effect=AssertionError("cache miss")):
+            assert await svc.count_user_states() == before
+        user.is_active = False
+        await finalize_session(db)
+        assert await svc.count_user_states() != before