From 0922cfd56d86a84a72d59e533b08e171dc9bec59 Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Thu, 5 Nov 2026 19:12:06 +0100 Subject: [PATCH 1/5] perf(users): off-loop password hashing everywhere; cached admin user summary; lighter admin users page Claude-Session: https://claude.ai/code/session_01M9neheZZEe3sVpDi2S3zT4 --- CHANGELOG.md | 8 + docs/modules/users.md | 15 ++ docs/perf/2026-11-05-users.md | 43 ++++ .../tests/test_users_admin_summary_cache.py | 185 ++++++++++++++++++ .../tests/test_users_password_offload.py | 122 ++++++++++++ modules/users/users/admin/queries.py | 29 +-- modules/users/users/admin/summary_cache.py | 131 +++++++++++++ modules/users/users/auth_local/token_api.py | 5 +- modules/users/users/core/__init__.py | 1 + modules/users/users/core/password_hashing.py | 61 ++++++ modules/users/users/db_adapter.py | 11 ++ modules/users/users/manager.py | 26 ++- modules/users/users/module.py | 2 + modules/users/users/roles_cache.py | 4 +- modules/users/users/startup.py | 2 + 15 files changed, 602 insertions(+), 43 deletions(-) create mode 100644 docs/perf/2026-11-05-users.md create mode 100644 modules/users/tests/test_users_admin_summary_cache.py create mode 100644 modules/users/tests/test_users_password_offload.py create mode 100644 modules/users/users/admin/summary_cache.py create mode 100644 modules/users/users/core/__init__.py create mode 100644 modules/users/users/core/password_hashing.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 12e20c5b9..071a6d292 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -41,6 +41,14 @@ All notable changes to this project are documented in this file. The format is b - `keycloak` follows the module layout: `keycloak.core.{constants,settings,state}`, `keycloak.identity.{provider,oidc,renewal,directory,persistence,tokens,jwks}` and `keycloak.web.endpoints`. `keycloak.{constants,settings,provider,jwks,tokens}` still import, with a `DeprecationWarning`, until 0.0.46. ### 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 cache, invalidated + after committed user, role and membership writes through the InvalidationBus. + List DTOs skip redundant validation of trusted database values; role options + select only the required columns, and managers share one password helper. - 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 diff --git a/docs/modules/users.md b/docs/modules/users.md index 19f4304a8..23d9f2792 100644 --- a/docs/modules/users.md +++ b/docs/modules/users.md @@ -14,6 +14,21 @@ The default auth provider + user-management module: email/password login, OAuth/ ## Auth flow +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 application for 30 seconds. Committed user, role and membership +writes invalidate them locally and publish through the InvalidationBus. With +Redis invalidation transport, other workers evict their entries too; without it, +or if a message is lost, another worker may show old totals until the TTL expires. +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](../performance/users.md) for the local +before/after results and reproduction command. + The module is built on [`fastapi-users`](https://fastapi-users.github.io/) for password hashing, registration, password reset, and verification. On top of that it layers: - A signed session cookie (`sm_auth` by default) — the auth module's `AuthMiddleware` calls `UsersAuthProvider.resolve_user`, which reads it on every request and populates `request.state.user`. Bearer tokens (`Authorization: Bearer `) resolve against `users_access_token`. diff --git a/docs/perf/2026-11-05-users.md b/docs/perf/2026-11-05-users.md new file mode 100644 index 000000000..ecd837e00 --- /dev/null +++ b/docs/perf/2026-11-05-users.md @@ -0,0 +1,43 @@ +# 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, user/role/membership writes, ORM and bulk DML, rollback, +remote invalidation, invalidation racing a cache fill, adapter-owned commits, +and role projections that do not instantiate ORM rows. DTO construction is +checked without invoking SQLModel input validation. diff --git a/modules/users/tests/test_users_admin_summary_cache.py b/modules/users/tests/test_users_admin_summary_cache.py new file mode 100644 index 000000000..1384779b0 --- /dev/null +++ b/modules/users/tests/test_users_admin_summary_cache.py @@ -0,0 +1,185 @@ +"""Admin summaries avoid repeated scans and forget only committed writes.""" + +from unittest.mock import patch + +import pytest +from fastapi_users.password import PasswordHelper +from simple_module_core.invalidation import Invalidation +from simple_module_db.transaction import finalize_session, rollback_session +from sqlalchemy import delete, event, select, update +from users.admin.service import UserService +from users.db_adapter import UserDatabaseWithRoles +from users.manager import UserManager +from users.models import Role, User, UserRole + + +def service(db, app): + manager = UserManager( + UserDatabaseWithRoles(db, User), app.state.users.mailer, app.state.users.settings + ) + return UserService(db, manager, app) + + +@pytest.mark.parametrize("method", ["count_user_states", "list_roles"]) +async def test_summary_cache_hit_and_expiry(users_app, monkeypatch, method): + from users.admin import summary_cache + + 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_committed_writes_invalidate_summaries(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": + from datetime import UTC, datetime + + user.invited_at = datetime.now(UTC) + elif write == "disable": + await db.execute(update(User).where(User.id == user.id).values(is_active=False)) + elif write == "delete": + await db.execute(delete(User).where(User.id == user.id)) + 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): + from users.admin.summary_cache import CHANNEL + + 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. + from users.admin import summary_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(CHANNEL, origin="other-worker").to_wire() + ) + # Invalidate remotely even when no local write occurred. + from users.admin import summary_cache + + with ( + patch.object(summary_cache, "load_counts", side_effect=RuntimeError("reloaded")), + pytest.raises(RuntimeError, match="reloaded"), + ): + await svc.count_user_states() + + +async def test_list_items_use_trusted_construction(users_app, monkeypatch): + from users.contracts.schemas import UserListItem + + 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] + + +@pytest.mark.parametrize("operation", ["create", "update", "delete"]) +async def test_adapter_commits_publish_before_request_finalization(users_app, operation): + from users.admin.summary_cache import CHANNEL + from users.contracts.schemas import UserCreate + + messages = [] + users_app.state.sm.invalidation.subscribe(CHANNEL, messages.append) + async with users_app.state.sm.db.session_factory() as db: + svc = service(db, users_app) + manager = svc._manager + await svc.count_user_states() + 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 messages + + +async def test_inflight_invalidation_does_not_repopulate_stale_cache(users_app, monkeypatch): + from users.admin import summary_cache + + 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): + from users.roles_cache import refresh_roles_cache + + 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) diff --git a/modules/users/tests/test_users_password_offload.py b/modules/users/tests/test_users_password_offload.py new file mode 100644 index 000000000..e39a00e06 --- /dev/null +++ b/modules/users/tests/test_users_password_offload.py @@ -0,0 +1,122 @@ +"""Crypto must run off-loop while the manager's DB operations stay on it.""" + +import asyncio +import os +import threading +import time +from unittest.mock import AsyncMock + +import pytest +from fastapi_users import exceptions +from users.admin.service import UserService +from users.contracts.schemas import UserCreate +from users.db_adapter import UserDatabaseWithRoles +from users.manager import UserManager +from users.models import User + + +def build_manager(db, app): + return UserManager( + UserDatabaseWithRoles(db, User), app.state.users.mailer, app.state.users.settings + ) + + +@pytest.mark.parametrize("operation", ["create", "forgot", "reset", "update"]) +async def test_crypto_runs_in_worker_thread(users_app, monkeypatch, operation): + async with users_app.state.sm.db.session_factory() as db: + manager = build_manager(db, users_app) + user = await manager.get_by_email("admin@example.com") + token = await manager.generate_reset_password_token(user) + loop_thread = threading.get_ident() + crypto_threads = [] + for name in ("hash", "verify_and_update"): + original = getattr(manager.password_helper, name) + + def record(*args, _original=original): + crypto_threads.append(threading.get_ident()) + return _original(*args) + + monkeypatch.setattr(manager.password_helper, name, record) + original_execute = db.execute + + async def execute(*args, **kwargs): + assert threading.get_ident() == loop_thread + return await original_execute(*args, **kwargs) + + monkeypatch.setattr(db, "execute", execute) + if operation == "create": + created = await manager.create( + UserCreate(email="new@example.com", password="SecurePass1!", is_verified=True) + ) + assert created.email == "new@example.com" + elif operation == "forgot": + manager.mailer.send_password_reset = AsyncMock() + await manager.forgot_password(user) + manager.mailer.send_password_reset.assert_awaited_once() + elif operation == "reset": + await manager.reset_password(token, "SecurePass1!") + with pytest.raises(exceptions.InvalidResetPasswordToken): + await manager.reset_password(token, "AnotherPass1!") + else: + await manager._update(user, {"password": "SecurePass1!"}) + assert crypto_threads + assert loop_thread not in crypto_threads + + +async def test_managers_share_password_helper(users_app): + async with users_app.state.sm.db.session_factory() as db: + assert ( + build_manager(db, users_app).password_helper + is build_manager(db, users_app).password_helper + ) + + +@pytest.mark.skipif( + not os.getenv("SM_TEST_REAL_PASSWORD_HASHING"), reason="opt-in production-cost measurement" +) +@pytest.mark.parametrize("operation", ["bulk_invite", "forgot"]) +async def test_real_hashing_loop_lag(users_app, operation): + """A 5 ms heartbeat records worst scheduler delay across real service work.""" + delays = [] + running = True + + async def heartbeat(): + while running: + started = time.perf_counter() + await asyncio.sleep(0.005) + delays.append(max(0, time.perf_counter() - started - 0.005)) + + ticker = asyncio.create_task(heartbeat()) + await asyncio.sleep(0) + started = time.perf_counter() + try: + async with users_app.state.sm.db.session_factory() as db: + manager = build_manager(db, users_app) + if operation == "bulk_invite": + service = UserService(db, manager, users_app) + for i in range(100): + await service.invite(f"lag-{i}@example.com", None, []) + else: + await manager.forgot_password(await manager.get_by_email("admin@example.com")) + finally: + running = False + await ticker + print( + f"{operation}: elapsed={time.perf_counter() - started:.3f}s " + f"max_loop_lag={max(delays) * 1000:.1f}ms" + ) + assert max(delays) < 0.05 + + +async def test_bearer_login_reuses_shared_helper(anon_client, monkeypatch): + from fastapi_users.password import PasswordHelper + + def reject_construction(*args, **kwargs): + raise AssertionError("request constructed another PasswordHelper") + + monkeypatch.setattr(PasswordHelper, "__init__", reject_construction) + response = await anon_client.post( + "/api/users/auth/token", + json={"email": "admin@example.com", "password": "AdminPass1!"}, + ) + assert response.status_code == 200, response.text diff --git a/modules/users/users/admin/queries.py b/modules/users/users/admin/queries.py index a478c3ac4..ef41a64e8 100644 --- a/modules/users/users/admin/queries.py +++ b/modules/users/users/admin/queries.py @@ -16,6 +16,7 @@ from sqlalchemy.orm import selectinload from sqlmodel import col +from users.admin.summary_cache import get_summary from users.admin.user_state import STATUS_CONDITIONS, user_state from users.contracts.schemas import RoleListItem, UserListItem from users.exceptions import EmailAlreadyExistsError, UserNotFoundError @@ -100,7 +101,7 @@ def _invite_expiry(self, invited_at: datetime | None) -> datetime | None: def to_list_item(self, user: User) -> UserListItem: """Build the DTO from a User with roles already eager-loaded.""" - return UserListItem( + return UserListItem.model_construct( id=user.id, email=user.email, full_name=user.full_name, @@ -118,22 +119,7 @@ def to_list_item(self, user: User) -> UserListItem: ) async def list_roles(self) -> list[RoleListItem]: - stmt = ( - select(Role, func.count(UserRole.user_id)) # ty: ignore[invalid-argument-type] - .outerjoin(UserRole, col(UserRole.role_id) == Role.id) - .group_by(Role.id) # ty: ignore[invalid-argument-type] - .order_by(Role.name) - ) - result = await self._db.execute(stmt) - return [ - RoleListItem( - id=role.id, - name=role.name, - description=role.description, - user_count=user_count, - ) - for role, user_count in result.all() - ] + return await get_summary(self._app, self._db, roles=True) async def _get_user_with_roles(self, user_id: uuid.UUID) -> User | None: # populate_existing: ``roles`` is ``noload``, so an instance already in @@ -259,7 +245,7 @@ async def list_users( # Selected columns == UserListItem fields (minus the derived ones and # roles): unpack directly. items = [ - UserListItem( + UserListItem.model_construct( **cast( "dict[str, Any]", {**r._mapping, "locked_until": _live_lock(r.locked_until)} ), @@ -284,12 +270,7 @@ async def count_user_states(self) -> dict[str, int]: One scan with ``COUNT(*) FILTER (WHERE …)`` per state rather than a round-trip each (Postgres and SQLite >= 3.30 both support it). """ - keys = ("active", "unverified", "invited") - stmt = select(*(func.count().filter(STATUS_CONDITIONS[k]()) for k in keys)).select_from( - User - ) - row = (await self._db.execute(stmt)).one() - return {key: int(value) for key, value in zip(keys, row, strict=True)} + return await get_summary(self._app, self._db, roles=False) async def get_with_roles(self, user_id: uuid.UUID) -> User | None: return await self._get_user_with_roles(user_id) diff --git a/modules/users/users/admin/summary_cache.py b/modules/users/users/admin/summary_cache.py new file mode 100644 index 000000000..f88ad865a --- /dev/null +++ b/modules/users/users/admin/summary_cache.py @@ -0,0 +1,131 @@ +"""Short-lived, app-local admin summaries, invalidated after user/role writes.""" + +from __future__ import annotations + +import time +from typing import Literal, overload + +from simple_module_db.callbacks import register_on_commit +from sqlalchemy import event, func, select +from sqlalchemy.ext.asyncio import async_session +from sqlmodel import col + +from users.admin.user_state import STATUS_CONDITIONS +from users.contracts.schemas import RoleListItem +from users.models import Role, User, UserRole + +CHANNEL = "users.admin_summary" +TTL_SECONDS = 30.0 +_TABLES = {User.__tablename__, Role.__tablename__, UserRole.__tablename__} + + +async def load_counts(db) -> dict[str, int]: + keys = ("active", "unverified", "invited") + stmt = select(*(func.count().filter(STATUS_CONDITIONS[k]()) for k in keys)).select_from(User) + row = (await db.execute(stmt)).one() + return {key: int(value) for key, value in zip(keys, row, strict=True)} + + +async def load_roles(db) -> list[RoleListItem]: + stmt = ( + select( + Role.id, + Role.name, + Role.description, + func.count(col(UserRole.user_id)).label("user_count"), + ) + .outerjoin(UserRole, col(UserRole.role_id) == Role.id) + .group_by(Role.id) + .order_by(Role.name) + ) + return [RoleListItem.model_construct(**row._mapping) for row in (await db.execute(stmt)).all()] + + +class SummaryCache: + def __init__(self): + self.slots = {} + self.generation = 0 + self.wired = False + + def forget(self): + self.slots.clear() + self.generation += 1 + + +@overload +async def get_summary(app, db, *, roles: Literal[True]) -> list[RoleListItem]: ... + + +@overload +async def get_summary(app, db, *, roles: Literal[False]) -> dict[str, int]: ... + + +async def get_summary(app, db, *, roles): + load = load_roles if roles else load_counts + # Service-only callers and writes in progress need their transaction's own view. + if app is None or db.new or db.dirty or db.deleted or db.info.get(CHANNEL): + return await load(db) + cache = app.state.users_admin_summary + slot = cache.slots.get(roles) + now = time.monotonic() + if slot is None or now >= slot[1]: + generation = cache.generation + # Never cache uncommitted rows or retain the request's transaction snapshot. + async with app.state.sm.db.session_factory() as committed_db: + value = await load(committed_db) + if generation == cache.generation: + cache.slots[roles] = (value, now + TTL_SECONDS) + else: + value = slot[0] + return [item.model_copy() for item in value] if roles else dict(value) + + +def subscribe(bus, app): + cache = SummaryCache() + app.state.users_admin_summary = cache + bus.subscribe(CHANNEL, lambda _: cache.forget()) + from users.roles_cache import CHANNEL as ROLES_CHANNEL + + bus.subscribe(ROLES_CHANNEL, lambda _: cache.forget()) + + +def register_writes(app): + cache = app.state.users_admin_summary + if cache.wired: + return + cache.wired = True + bus = app.state.sm.invalidation + + def written(session): + session.info[CHANNEL] = True + db = async_session(session) + assert db is not None + register_on_commit(db, lambda: bus.publish(CHANNEL)) + + def before_flush(session, *_): + if any( + isinstance(obj, User | Role | UserRole) + for obj in (*session.new, *session.dirty, *session.deleted) + ): + written(session) + + def before_execute(state): + if (state.is_insert or state.is_update or state.is_delete) and ( + getattr(getattr(state.statement, "table", None), "name", None) in _TABLES + ): + written(state.session) + + def after_commit(session): + if not session.in_nested_transaction() and session.info.pop(CHANNEL, False): + # Also covers fastapi-users adapters that commit before request finalization. + cache.forget() + + def after_rollback(session, transaction): + if transaction.parent is None: + session.info.pop(CHANNEL, None) + + session_class = app.state.sm.db.sync_session_class + event.listen(session_class, "before_flush", before_flush) + event.listen(session_class, "do_orm_execute", before_execute) + event.listen(session_class, "after_commit", after_commit) + event.listen(session_class, "after_soft_rollback", after_rollback) diff --git a/modules/users/users/auth_local/token_api.py b/modules/users/users/auth_local/token_api.py index c1086b068..9c7217b69 100644 --- a/modules/users/users/auth_local/token_api.py +++ b/modules/users/users/auth_local/token_api.py @@ -26,6 +26,7 @@ record_success, release, ) +from users.core.password_hashing import PASSWORD_HELPER from users.demo.guard import DEMO_READ_ONLY_DETAIL, is_read_only_demo from users.models import User from users.models.refresh_token import RefreshToken @@ -68,9 +69,7 @@ async def token_login( db: AsyncSession = Depends(get_db), ): """Exchange email + password for an access/refresh token pair.""" - from fastapi_users.password import PasswordHelper - - helper = PasswordHelper() + helper = PASSWORD_HELPER # Same lockout as the session login (GH #530): a slot is taken before the # password is checked, and a success gives it back. attempt = login_attempt(request, body.email) diff --git a/modules/users/users/core/__init__.py b/modules/users/users/core/__init__.py new file mode 100644 index 000000000..f002e1469 --- /dev/null +++ b/modules/users/users/core/__init__.py @@ -0,0 +1 @@ +"""Users domain core: framework-free helpers.""" diff --git a/modules/users/users/core/password_hashing.py b/modules/users/users/core/password_hashing.py new file mode 100644 index 000000000..429251194 --- /dev/null +++ b/modules/users/users/core/password_hashing.py @@ -0,0 +1,61 @@ +"""Async crypto boundaries for fastapi-users 15's synchronous password helper.""" + +from __future__ import annotations + +import asyncio + +import jwt +from fastapi_users import exceptions +from fastapi_users.jwt import decode_jwt +from fastapi_users.password import PasswordHelper + +PASSWORD_HELPER = PasswordHelper() + + +class PasswordHashingMixin: + """Keep validation, database operations and lifecycle hooks on the event loop.""" + + async def create(self, user_create, safe=False, request=None): + await self.validate_password(user_create.password, user_create) + if await self.user_db.get_by_email(user_create.email) is not None: + raise exceptions.UserAlreadyExists() + data = ( + user_create.create_update_dict() if safe else user_create.create_update_dict_superuser() + ) + password = data.pop("password") + data["hashed_password"] = await asyncio.to_thread(self.password_helper.hash, password) + user = await self.user_db.create(data) + await self.on_after_register(user, request) + return user + + async def _update(self, user, update_dict): + # Let upstream handle email uniqueness and every non-password field. + data = dict(update_dict) + if data.get("password") is not None: + password = data.pop("password") + await self.validate_password(password, user) + data["hashed_password"] = await asyncio.to_thread(self.password_helper.hash, password) + return await super()._update(user, data) + + async def reset_password(self, token, password, request=None): + # Mirrors upstream token/fingerprint checks; only crypto leaves the loop. + try: + data = decode_jwt( + token, self.reset_password_token_secret, [self.reset_password_token_audience] + ) + user_id = data["sub"] + fingerprint = data["password_fgpt"] + parsed_id = self.parse_id(user_id) + except (jwt.PyJWTError, KeyError, exceptions.InvalidID): + raise exceptions.InvalidResetPasswordToken() from None + user = await self.get(parsed_id) + valid, _ = await asyncio.to_thread( + self.password_helper.verify_and_update, user.hashed_password, fingerprint + ) + if not valid: + raise exceptions.InvalidResetPasswordToken() + if not user.is_active: + raise exceptions.UserInactive() + updated = await self._update(user, {"password": password}) + await self.on_after_reset_password(updated, request) + return updated diff --git a/modules/users/users/db_adapter.py b/modules/users/users/db_adapter.py index 5b8f5099a..b9e9939dd 100644 --- a/modules/users/users/db_adapter.py +++ b/modules/users/users/db_adapter.py @@ -9,6 +9,7 @@ from fastapi_users.exceptions import UserAlreadyExists from fastapi_users_db_sqlalchemy import SQLAlchemyUserDatabase from fastapi_users_db_sqlalchemy.access_token import SQLAlchemyAccessTokenDatabase +from simple_module_db.callbacks import run_on_commit_callbacks from simple_module_db.deps import get_db from sqlalchemy import func, select from sqlalchemy.exc import IntegrityError @@ -48,9 +49,19 @@ async def create(self, create_dict): raise UserAlreadyExists() from None raise await self.session.commit() + await run_on_commit_callbacks(self.session) await self.session.refresh(user) return user + async def update(self, user, update_dict): + user = await super().update(user, update_dict) + await run_on_commit_callbacks(self.session) + return user + + async def delete(self, user): + await super().delete(user) + await run_on_commit_callbacks(self.session) + async def get(self, id): stmt = ( select(self.user_table) diff --git a/modules/users/users/manager.py b/modules/users/users/manager.py index 0891b3a2d..99f077360 100644 --- a/modules/users/users/manager.py +++ b/modules/users/users/manager.py @@ -21,6 +21,7 @@ SESSION_VERSION_KEY, ) from users.contracts.events import UserRegistered +from users.core.password_hashing import PASSWORD_HELPER, PasswordHashingMixin from users.db_adapter import UserDatabaseWithRoles, get_user_db from users.exceptions import ExternalUserNoPasswordError from users.mailer import Mailer @@ -34,7 +35,9 @@ logger = logging.getLogger(__name__) -class UserManager(ResetRevocationMixin, UUIDIDMixin, BaseUserManager[User, uuid.UUID]): +class UserManager( + ResetRevocationMixin, PasswordHashingMixin, UUIDIDMixin, BaseUserManager[User, uuid.UUID] +): """Customizes password validation, token secrets, and lifecycle emails.""" # Secrets and lifetimes pulled from settings at construction time so @@ -45,7 +48,7 @@ def __init__( mailer: Mailer, settings: UsersSettings, ) -> None: - super().__init__(user_db) + super().__init__(user_db, PASSWORD_HELPER) self.mailer = mailer self.reset_password_token_secret = settings.reset_password_token_secret self.verification_token_secret = settings.verification_token_secret @@ -111,7 +114,10 @@ async def forgot_password(self, user: User, request: Request | None = None) -> N """ if user.is_external or user.hashed_password is None: return - await super().forgot_password(user, request) + if not user.is_active: + raise exceptions.UserInactive() + token = await self.generate_reset_password_token(user) + await self.on_after_forgot_password(user, token, request) # ── Lifecycle hooks ────────────────────────────────────── @@ -258,17 +264,9 @@ def mint_invite_token(self, user: User, invited_by: str | None = None) -> str: async def generate_reset_password_token(self, user: User) -> str: """Mint a reset-audience JWT without firing on_after_forgot_password. - Shape matches ``BaseUserManager.forgot_password``: the payload includes - ``password_fgpt`` (a bcrypt of the current hashed_password) so the token - invalidates when the password changes. Used by the admin - reset-password-link endpoint, where the admin copies the link instead - of triggering an email. - - The fingerprint must be bcrypt-style because the stock - ``reset_password`` path in fastapi-users verifies it with - ``password_helper.verify_and_update``. Bcrypt is CPU-bound (~100ms at - default rounds) so we offload it to a worker thread — otherwise a - single admin action would stall the event loop for other requests. + The payload fingerprints the current hashed password, so changing the + password invalidates the token. The CPU-bound fingerprint is hashed in + a worker thread for both forgot-password mail and admin reset links. """ if user.is_external or user.hashed_password is None: # No local password to fingerprint — hashing None would crash. diff --git a/modules/users/users/module.py b/modules/users/users/module.py index 236210b22..6e35f0652 100644 --- a/modules/users/users/module.py +++ b/modules/users/users/module.py @@ -126,12 +126,14 @@ def register_invalidations(self, bus: InvalidationBus, app: FastAPI) -> None: sessions for the rest of their TTL (GH #318). """ from users import roles_cache + from users.admin import summary_cache from users.session_revocation import subscribe subscribe(bus) # The admin user screens' cached role list: the setup wizard can add # the admin role after every worker cached the seeded ``user`` role. roles_cache.subscribe(bus, app) + summary_cache.subscribe(bus, app) def register_exception_handlers(self, app: FastAPI) -> None: from users.errors import install_exception_handlers diff --git a/modules/users/users/roles_cache.py b/modules/users/users/roles_cache.py index 5fbb00ee2..843b3c41b 100644 --- a/modules/users/users/roles_cache.py +++ b/modules/users/users/roles_cache.py @@ -52,8 +52,8 @@ class RoleSummary: async def refresh_roles_cache(app: FastAPI) -> list[RoleSummary]: """Reload the roles list from the DB into ``app.state.users.roles_cache``.""" async with app.state.sm.db.session_factory() as db: - result = await db.execute(select(Role).order_by(Role.name)) - cached = [RoleSummary(id=str(r.id), name=r.name) for r in result.scalars().all()] + result = await db.execute(select(Role.id, Role.name).order_by(Role.name)) + cached = [RoleSummary(id=str(r.id), name=r.name) for r in result.all()] app.state.users.roles_cache = cached app.state.users.roles_cache_loaded_at = time.monotonic() return cached diff --git a/modules/users/users/startup.py b/modules/users/users/startup.py index 0b00a4748..dc6bb45fe 100644 --- a/modules/users/users/startup.py +++ b/modules/users/users/startup.py @@ -44,6 +44,7 @@ async def run_startup(module: UsersModule, app: FastAPI) -> None: """Populate ``app.state.users`` and apply settings-derived configuration.""" + from users.admin.summary_cache import register_writes from users.backend import reconfigure_cookie_transport from users.bootstrap import bootstrap_admin_from_env, ensure_user_role from users.demo import ensure_demo_users @@ -52,6 +53,7 @@ async def run_startup(module: UsersModule, app: FastAPI) -> None: from users.session_version_cache import configure_session_version_cache from users.settings import DEFAULT_LOGIN_REDIRECT_URL + register_writes(app) state = app.state.users s = state.settings From a7e7cbc8173d69ba243da26203bf194d3a877437 Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Thu, 5 Nov 2026 19:17:40 +0100 Subject: [PATCH 2/5] fix(users): admin summary invalidates only on counted-state writes, once per transaction; lazy cache Claude-Session: https://claude.ai/code/session_01M9neheZZEe3sVpDi2S3zT4 --- CHANGELOG.md | 7 +- docs/modules/users.md | 12 ++- .../tests/test_users_admin_summary_cache.py | 60 ++++++++++++ modules/users/users/admin/service.py | 4 +- modules/users/users/admin/summary_cache.py | 97 +++++++++++++++---- 5 files changed, 155 insertions(+), 25 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 071a6d292..f61fe5a28 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -45,8 +45,11 @@ All notable changes to this project are documented in this file. The format is b 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 cache, invalidated - after committed user, role and membership writes through the InvalidationBus. +- Admin user state totals and role options use a 30-second cache. Committed + writes that change a counted state (user created, deleted, enabled, verified + or invited; role or membership changes) invalidate it and publish one + `users.admin_summary` message per transaction; logins, lockout counters and + password changes publish nothing. List DTOs skip redundant validation of trusted database values; role options select only the required columns, and managers share one password helper. diff --git a/docs/modules/users.md b/docs/modules/users.md index 23d9f2792..403d58348 100644 --- a/docs/modules/users.md +++ b/docs/modules/users.md @@ -20,10 +20,14 @@ 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 application for 30 seconds. Committed user, role and membership -writes invalidate them locally and publish through the InvalidationBus. With -Redis invalidation transport, other workers evict their entries too; without it, -or if a message is lost, another worker may show old totals until the TTL expires. +cached per application for 30 seconds. Only committed writes that can move a +figure invalidate them: a user created, deleted, enabled/disabled, verified or +invited; a role created, deleted or renamed; a membership added or removed. +Such a transaction evicts the local entry and publishes one `users.admin_summary` +message through the InvalidationBus. Logins, failed-login counters, password +changes and session revocations publish nothing. With the Redis invalidation +transport, other workers evict their entries too; without it, or if a message is +lost, another worker may show old totals until the TTL expires. 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](../performance/users.md) for the local diff --git a/modules/users/tests/test_users_admin_summary_cache.py b/modules/users/tests/test_users_admin_summary_cache.py index 1384779b0..0acefab1f 100644 --- a/modules/users/tests/test_users_admin_summary_cache.py +++ b/modules/users/tests/test_users_admin_summary_cache.py @@ -1,5 +1,6 @@ """Admin summaries avoid repeated scans and forget only committed writes.""" +from datetime import UTC, datetime from unittest.mock import patch import pytest @@ -7,6 +8,7 @@ from simple_module_core.invalidation import Invalidation from simple_module_db.transaction import finalize_session, rollback_session from sqlalchemy import delete, event, select, update +from users.admin import summary_cache from users.admin.service import UserService from users.db_adapter import UserDatabaseWithRoles from users.manager import UserManager @@ -183,3 +185,61 @@ def record(*args): assert loaded == [] finally: event.remove(Role, "load", record) + + +def _record(app): + from users.admin.summary_cache import CHANNEL + + messages = [] + app.state.sm.invalidation.subscribe(CHANNEL, messages.append) + return messages + + +async def test_uncounted_user_writes_publish_nothing(users_app): + """Logins, lockout counters and revocations must not chatter on the bus.""" + messages = _record(users_app) + async with users_app.state.sm.db.session_factory() as db: + svc = service(db, users_app) + before = await svc.count_user_states() + user = ( + await db.execute(select(User).where(User.email == "admin@example.com")) + ).scalar_one() + user.session_version = int(user.session_version or 0) + 1 + user.hashed_password = "rotated" + user.last_login_at = datetime.now(UTC) + await db.flush() + await db.execute( + update(User).where(User.id == user.id).values(failed_login_count=3, session_version=99) + ) + await finalize_session(db) + with patch.object( + summary_cache, "load_counts", side_effect=AssertionError("cache dropped") + ): + assert await svc.count_user_states() == before + assert messages == [] + + +async def test_one_publish_per_transaction(users_app): + messages = _record(users_app) + async with users_app.state.sm.db.session_factory() as db: + db.add(User(email="one@example.com", hashed_password="unused")) + await db.flush() + db.add(User(email="two@example.com", hashed_password="unused")) + await db.flush() + await db.execute( + update(User).where(User.email == "one@example.com").values(is_active=False) + ) + await finalize_session(db) + assert len(messages) == 1 + + +async def test_savepoint_rollback_keeps_a_later_write_publishing(users_app): + messages = _record(users_app) + async with users_app.state.sm.db.session_factory() as db: + nested = await db.begin_nested() + db.add(User(email="undone@example.com", hashed_password="unused")) + await db.flush() + await nested.rollback() + db.add(User(email="kept@example.com", hashed_password="unused")) + await finalize_session(db) + assert len(messages) == 1 diff --git a/modules/users/users/admin/service.py b/modules/users/users/admin/service.py index 1abe19d15..ba1767e94 100644 --- a/modules/users/users/admin/service.py +++ b/modules/users/users/admin/service.py @@ -57,7 +57,7 @@ async def create_user( is_verified=True, is_superuser=False, ) - user = await self._manager.create(user_create, safe=False) # ty: ignore[invalid-argument-type] + user = await self._manager.create(user_create, safe=False) roles = await self._resolve_roles(role_names) for role in roles: @@ -159,7 +159,7 @@ async def invite( is_active=True, is_verified=False, ) - user = await self._manager.create(user_create, safe=False) # ty: ignore[invalid-argument-type] + user = await self._manager.create(user_create, safe=False) user.invited_at = datetime.now(UTC) # Assign roles diff --git a/modules/users/users/admin/summary_cache.py b/modules/users/users/admin/summary_cache.py index f88ad865a..c2f2478b6 100644 --- a/modules/users/users/admin/summary_cache.py +++ b/modules/users/users/admin/summary_cache.py @@ -1,4 +1,14 @@ -"""Short-lived, app-local admin summaries, invalidated after user/role writes.""" +"""Short-lived, app-local admin summaries, invalidated after user/role writes. + +Only writes that can change a summary invalidate it: a user created, deleted, +enabled/disabled, verified or invited; a role created, deleted or renamed; a +membership added or removed. Logins, failed-login counters, password changes +and session revocations touch ``users_user`` too but change no figure here, so +they publish nothing. A qualifying transaction publishes at most once, after +commit. Without a cross-process transport (or if a message is lost) another +worker shows old totals until :data:`TTL_SECONDS` expires; these figures are +informational and never gate access. +""" from __future__ import annotations @@ -6,7 +16,7 @@ from typing import Literal, overload from simple_module_db.callbacks import register_on_commit -from sqlalchemy import event, func, select +from sqlalchemy import event, func, inspect, select from sqlalchemy.ext.asyncio import async_session from sqlmodel import col @@ -16,7 +26,13 @@ CHANNEL = "users.admin_summary" TTL_SECONDS = 30.0 -_TABLES = {User.__tablename__, Role.__tablename__, UserRole.__tablename__} +# Columns whose change can move a summary figure; ``None`` means any change. +_COUNTED: dict[str, frozenset[str] | None] = { + User.__tablename__: frozenset({"is_active", "is_verified", "invited_at"}), + Role.__tablename__: frozenset({"name", "description"}), + UserRole.__tablename__: None, +} +_TOKEN = f"{CHANNEL}.publish" async def load_counts(db) -> dict[str, int]: @@ -52,6 +68,44 @@ def forget(self): self.generation += 1 +def _cache(app) -> SummaryCache: + """The app's cache, created on first use. + + Lazy so ``register_invalidations`` only subscribes: it must not assume an + app it can write state onto (the hook is also called app-less). + """ + cache = getattr(app.state, "users_admin_summary", None) + if cache is None: + cache = app.state.users_admin_summary = SummaryCache() + return cache + + +def _changes_summary(obj, dirty: bool) -> bool: + if not isinstance(obj, User | Role | UserRole): + return False + columns = _COUNTED[obj.__tablename__] + if not dirty or columns is None: + return True + attrs = inspect(obj).attrs + return any(attrs[name].history.has_changes() for name in columns) + + +def _statement_changes_summary(state) -> bool: + if not (state.is_insert or state.is_update or state.is_delete): + return False + name = getattr(getattr(state.statement, "table", None), "name", None) + if name not in _COUNTED: + return False + columns = _COUNTED[name] + if not state.is_update or columns is None: + return True + values = getattr(state.statement, "_values", None) + if not values: + # Bulk-by-primary-key or ordered values: unknown columns, stay safe. + return True + return any(getattr(key, "key", key) in columns for key in values) + + @overload async def get_summary(app, db, *, roles: Literal[True]) -> list[RoleListItem]: ... @@ -65,7 +119,7 @@ async def get_summary(app, db, *, roles): # Service-only callers and writes in progress need their transaction's own view. if app is None or db.new or db.dirty or db.deleted or db.info.get(CHANNEL): return await load(db) - cache = app.state.users_admin_summary + cache = _cache(app) slot = cache.slots.get(roles) now = time.monotonic() if slot is None or now >= slot[1]: @@ -81,16 +135,14 @@ async def get_summary(app, db, *, roles): def subscribe(bus, app): - cache = SummaryCache() - app.state.users_admin_summary = cache - bus.subscribe(CHANNEL, lambda _: cache.forget()) from users.roles_cache import CHANNEL as ROLES_CHANNEL - bus.subscribe(ROLES_CHANNEL, lambda _: cache.forget()) + bus.subscribe(CHANNEL, lambda _: _cache(app).forget()) + bus.subscribe(ROLES_CHANNEL, lambda _: _cache(app).forget()) def register_writes(app): - cache = app.state.users_admin_summary + cache = _cache(app) if cache.wired: return cache.wired = True @@ -98,31 +150,42 @@ def register_writes(app): def written(session): session.info[CHANNEL] = True + # One shared token per transaction: every qualifying flush queues a + # callback (so a savepoint rollback cannot drop the only one), but the + # first to run publishes and the rest are no-ops. + token = session.info.setdefault(_TOKEN, {"sent": False}) + + async def publish(): + if not token["sent"]: + token["sent"] = True + await bus.publish(CHANNEL) + db = async_session(session) assert db is not None - register_on_commit(db, lambda: bus.publish(CHANNEL)) + register_on_commit(db, publish) def before_flush(session, *_): - if any( - isinstance(obj, User | Role | UserRole) - for obj in (*session.new, *session.dirty, *session.deleted) + if any(_changes_summary(obj, False) for obj in (*session.new, *session.deleted)) or any( + _changes_summary(obj, True) for obj in session.dirty ): written(session) def before_execute(state): - if (state.is_insert or state.is_update or state.is_delete) and ( - getattr(getattr(state.statement, "table", None), "name", None) in _TABLES - ): + if _statement_changes_summary(state): written(state.session) def after_commit(session): - if not session.in_nested_transaction() and session.info.pop(CHANNEL, False): + if session.in_nested_transaction(): + return + session.info.pop(_TOKEN, None) + if session.info.pop(CHANNEL, False): # Also covers fastapi-users adapters that commit before request finalization. cache.forget() def after_rollback(session, transaction): if transaction.parent is None: session.info.pop(CHANNEL, None) + session.info.pop(_TOKEN, None) session_class = app.state.sm.db.sync_session_class event.listen(session_class, "before_flush", before_flush) From 8ec6f11a266273c8f9dc39aae03b7059e1e1d9cc Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Thu, 5 Nov 2026 19:26:25 +0100 Subject: [PATCH 3/5] refactor(optimize): admin summary cache uses local commit eviction + explicit publishes from admin writes Claude-Session: https://claude.ai/code/session_01M9neheZZEe3sVpDi2S3zT4 --- CHANGELOG.md | 15 +- docs/modules/users.md | 19 +- docs/perf/2026-11-05-users.md | 10 +- .../tests/test_users_admin_summary_cache.py | 186 +++++++++--------- modules/users/users/admin/queries.py | 10 +- modules/users/users/admin/service.py | 37 ++-- modules/users/users/admin/summary_cache.py | 186 +++++++----------- modules/users/users/auth_local/token_api.py | 5 +- modules/users/users/core/__init__.py | 2 +- modules/users/users/core/password_hashing.py | 18 +- 10 files changed, 225 insertions(+), 263 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 51e891f0b..3e454159f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -46,14 +46,17 @@ All notable changes to this project are documented in this file. The format is b 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 cache. Committed - writes that change a counted state (user created, deleted, enabled, verified - or invited; role or membership changes) invalidate it and publish one - `users.admin_summary` message per transaction; logins, lockout counters and - password changes publish nothing. +- 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 diff --git a/docs/modules/users.md b/docs/modules/users.md index 403d58348..1c166f11a 100644 --- a/docs/modules/users.md +++ b/docs/modules/users.md @@ -20,17 +20,18 @@ 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 application for 30 seconds. Only committed writes that can move a -figure invalidate them: a user created, deleted, enabled/disabled, verified or -invited; a role created, deleted or renamed; a membership added or removed. -Such a transaction evicts the local entry and publishes one `users.admin_summary` -message through the InvalidationBus. Logins, failed-login counters, password -changes and session revocations publish nothing. With the Redis invalidation -transport, other workers evict their entries too; without it, or if a message is -lost, another worker may show old totals until the TTL expires. +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](../performance/users.md) for the local +See [Users performance measurements](../perf/2026-11-05-users.md) for the local before/after results and reproduction command. The module is built on [`fastapi-users`](https://fastapi-users.github.io/) for password hashing, registration, password reset, and verification. On top of that it layers: diff --git a/docs/perf/2026-11-05-users.md b/docs/perf/2026-11-05-users.md index ecd837e00..3640783fb 100644 --- a/docs/perf/2026-11-05-users.md +++ b/docs/perf/2026-11-05-users.md @@ -37,7 +37,9 @@ 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, user/role/membership writes, ORM and bulk DML, rollback, -remote invalidation, invalidation racing a cache fill, adapter-owned commits, -and role projections that do not instantiate ORM rows. DTO construction is -checked without invoking SQLModel input validation. +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. diff --git a/modules/users/tests/test_users_admin_summary_cache.py b/modules/users/tests/test_users_admin_summary_cache.py index 0acefab1f..741c9b191 100644 --- a/modules/users/tests/test_users_admin_summary_cache.py +++ b/modules/users/tests/test_users_admin_summary_cache.py @@ -1,18 +1,22 @@ -"""Admin summaries avoid repeated scans and forget only committed writes.""" +"""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 delete, event, select, update +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): @@ -22,10 +26,22 @@ def service(db, app): 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): - from users.admin import summary_cache - queries = [] event.listen( users_app.state.sm.db.engine.sync_engine, @@ -49,7 +65,7 @@ async def test_summary_cache_hit_and_expiry(users_app, monkeypatch, method): @pytest.mark.parametrize( "write", ["create", "verify", "invite", "disable", "delete", "role", "membership"] ) -async def test_committed_writes_invalidate_summaries(users_app, write): +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( @@ -64,13 +80,11 @@ async def test_committed_writes_invalidate_summaries(users_app, write): elif write == "verify": user.is_verified = True elif write == "invite": - from datetime import UTC, datetime - user.invited_at = datetime.now(UTC) elif write == "disable": - await db.execute(update(User).where(User.id == user.id).values(is_active=False)) + user.is_active = False elif write == "delete": - await db.execute(delete(User).where(User.id == user.id)) + await db.delete(user) elif write == "role": db.add(Role(name="reviewer")) else: @@ -84,8 +98,6 @@ async def test_committed_writes_invalidate_summaries(users_app, write): async def test_rollback_and_remote_invalidation(users_app): - from users.admin.summary_cache import CHANNEL - async with users_app.state.sm.db.session_factory() as db: svc = service(db, users_app) before = await svc.count_user_states() @@ -93,16 +105,12 @@ async def test_rollback_and_remote_invalidation(users_app): await db.flush() await rollback_session(db) # A rollback must preserve a warm cache. - from users.admin import summary_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(CHANNEL, origin="other-worker").to_wire() + Invalidation(summary_cache.CHANNEL, origin="other-worker").to_wire() ) # Invalidate remotely even when no local write occurred. - from users.admin import summary_cache - with ( patch.object(summary_cache, "load_counts", side_effect=RuntimeError("reloaded")), pytest.raises(RuntimeError, match="reloaded"), @@ -110,32 +118,63 @@ async def test_rollback_and_remote_invalidation(users_app): await svc.count_user_states() -async def test_list_items_use_trusted_construction(users_app, monkeypatch): - from users.contracts.schemas import UserListItem +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) - def reject_validation(*args, **kwargs): - raise AssertionError("DB rows should not be validated again") - monkeypatch.setattr(UserListItem, "__init__", reject_validation) +_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: - 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] + await _call(service(db, users_app), method, user_id) + assert messages == [] + await finalize_session(db) + assert len(messages) == 1 -@pytest.mark.parametrize("operation", ["create", "update", "delete"]) -async def test_adapter_commits_publish_before_request_finalization(users_app, operation): - from users.admin.summary_cache import CHANNEL - from users.contracts.schemas import UserCreate +@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 == [] - messages = [] - users_app.state.sm.invalidation.subscribe(CHANNEL, messages.append) + +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: - svc = service(db, users_app) - manager = svc._manager - await svc.count_user_states() + 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) @@ -146,12 +185,23 @@ async def test_adapter_commits_publish_before_request_finalization(users_app, op await manager._update(user, {"is_active": False}) else: await manager.delete(user) - assert messages + assert ran == [operation] -async def test_inflight_invalidation_does_not_repopulate_stale_cache(users_app, monkeypatch): - from users.admin import summary_cache +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): @@ -170,8 +220,6 @@ async def racing_load(db): async def test_role_options_do_not_hydrate_orm_rows(users_app): - from users.roles_cache import refresh_roles_cache - loaded = [] def record(*args): @@ -185,61 +233,3 @@ def record(*args): assert loaded == [] finally: event.remove(Role, "load", record) - - -def _record(app): - from users.admin.summary_cache import CHANNEL - - messages = [] - app.state.sm.invalidation.subscribe(CHANNEL, messages.append) - return messages - - -async def test_uncounted_user_writes_publish_nothing(users_app): - """Logins, lockout counters and revocations must not chatter on the bus.""" - messages = _record(users_app) - async with users_app.state.sm.db.session_factory() as db: - svc = service(db, users_app) - before = await svc.count_user_states() - user = ( - await db.execute(select(User).where(User.email == "admin@example.com")) - ).scalar_one() - user.session_version = int(user.session_version or 0) + 1 - user.hashed_password = "rotated" - user.last_login_at = datetime.now(UTC) - await db.flush() - await db.execute( - update(User).where(User.id == user.id).values(failed_login_count=3, session_version=99) - ) - await finalize_session(db) - with patch.object( - summary_cache, "load_counts", side_effect=AssertionError("cache dropped") - ): - assert await svc.count_user_states() == before - assert messages == [] - - -async def test_one_publish_per_transaction(users_app): - messages = _record(users_app) - async with users_app.state.sm.db.session_factory() as db: - db.add(User(email="one@example.com", hashed_password="unused")) - await db.flush() - db.add(User(email="two@example.com", hashed_password="unused")) - await db.flush() - await db.execute( - update(User).where(User.email == "one@example.com").values(is_active=False) - ) - await finalize_session(db) - assert len(messages) == 1 - - -async def test_savepoint_rollback_keeps_a_later_write_publishing(users_app): - messages = _record(users_app) - async with users_app.state.sm.db.session_factory() as db: - nested = await db.begin_nested() - db.add(User(email="undone@example.com", hashed_password="unused")) - await db.flush() - await nested.rollback() - db.add(User(email="kept@example.com", hashed_password="unused")) - await finalize_session(db) - assert len(messages) == 1 diff --git a/modules/users/users/admin/queries.py b/modules/users/users/admin/queries.py index ef41a64e8..580661f9d 100644 --- a/modules/users/users/admin/queries.py +++ b/modules/users/users/admin/queries.py @@ -16,7 +16,7 @@ from sqlalchemy.orm import selectinload from sqlmodel import col -from users.admin.summary_cache import get_summary +from users.admin.summary_cache import get_role_summary, get_state_counts from users.admin.user_state import STATUS_CONDITIONS, user_state from users.contracts.schemas import RoleListItem, UserListItem from users.exceptions import EmailAlreadyExistsError, UserNotFoundError @@ -119,7 +119,7 @@ def to_list_item(self, user: User) -> UserListItem: ) async def list_roles(self) -> list[RoleListItem]: - return await get_summary(self._app, self._db, roles=True) + return await get_role_summary(self._app, self._db) async def _get_user_with_roles(self, user_id: uuid.UUID) -> User | None: # populate_existing: ``roles`` is ``noload``, so an instance already in @@ -267,10 +267,10 @@ async def count_user_states(self) -> dict[str, int]: self-registered account that never clicked its verification mail is nobody's to chase. - One scan with ``COUNT(*) FILTER (WHERE …)`` per state rather than a - round-trip each (Postgres and SQLite >= 3.30 both support it). + Cached for 30 seconds per worker (``summary_cache.TTL_SECONDS``); + admin writes evict it on commit. """ - return await get_summary(self._app, self._db, roles=False) + return await get_state_counts(self._app, self._db) async def get_with_roles(self, user_id: uuid.UUID) -> User | None: return await self._get_user_with_roles(user_id) diff --git a/modules/users/users/admin/service.py b/modules/users/users/admin/service.py index ba1767e94..1d986a202 100644 --- a/modules/users/users/admin/service.py +++ b/modules/users/users/admin/service.py @@ -11,6 +11,7 @@ from sqlmodel import col from users.admin.safety import _UserAdminSafety +from users.admin.summary_cache import publish_after_commit from users.constants import ADMIN_ROLE_NAME from users.contracts.schemas import UserCreate from users.exceptions import ( @@ -58,16 +59,12 @@ async def create_user( is_superuser=False, ) user = await self._manager.create(user_create, safe=False) + # Queued after the adapter's own commit, so it waits for the roles too. + publish_after_commit(self._app, self._db) roles = await self._resolve_roles(role_names) for role in roles: - self._db.add( - UserRole( - user_id=user.id, - role_id=role.id, - assigned_by=created_by, - ) - ) + self._db.add(UserRole(user_id=user.id, role_id=role.id, assigned_by=created_by)) if roles: await self._db.flush() # User.roles is lazy="noload": selectinload only populates a fresh @@ -134,6 +131,7 @@ async def delete_user(self, user_id: uuid.UUID) -> None: await self._db.execute(delete(model).where(col(model.user_id) == user_id)) await self._db.delete(user) await self._db.flush() + publish_after_commit(self._app, self._db) async def invite( self, @@ -161,18 +159,12 @@ async def invite( ) user = await self._manager.create(user_create, safe=False) user.invited_at = datetime.now(UTC) + publish_after_commit(self._app, self._db) - # Assign roles roles = await self._resolve_roles(role_names) invited_by_str = str(invited_by.id) if invited_by else None for role in roles: - self._db.add( - UserRole( - user_id=user.id, - role_id=role.id, - assigned_by=invited_by_str, - ) - ) + self._db.add(UserRole(user_id=user.id, role_id=role.id, assigned_by=invited_by_str)) await self._db.flush() if roles: await self._db.refresh(user, attribute_names=["roles"]) @@ -224,6 +216,7 @@ async def disable(self, user_id: uuid.UUID) -> User: .values(revoked_at=datetime.now(UTC)) ) await self._db.flush() + publish_after_commit(self._app, self._db) return user async def enable(self, user_id: uuid.UUID) -> User: @@ -232,6 +225,7 @@ async def enable(self, user_id: uuid.UUID) -> User: user.is_active = True await self._revoke_sessions(user) await self._db.flush() + publish_after_commit(self._app, self._db) return user async def unlock(self, user_id: uuid.UUID) -> User: @@ -248,6 +242,7 @@ async def mark_verified(self, user_id: uuid.UUID) -> User: if not user.is_verified: user.is_verified = True await self._db.flush() + publish_after_commit(self._app, self._db) return user async def set_roles( @@ -262,22 +257,16 @@ async def set_roles( if ADMIN_ROLE_NAME in {r.name for r in user.roles} and ADMIN_ROLE_NAME not in role_names: await self._guard_admin_removal(user) - # Delete all existing role assignments for this user await self._db.execute(delete(UserRole).where(col(UserRole.user_id) == user_id)) - # Insert new role assignments roles = await self._resolve_roles(role_names) for role in roles: - self._db.add( - UserRole( - user_id=user_id, - role_id=role.id, - assigned_by=assigned_by, - ) - ) + self._db.add(UserRole(user_id=user_id, role_id=role.id, assigned_by=assigned_by)) await self._revoke_sessions(user) await self._db.flush() + # The bulk DELETE above is Core DML, which the local flush marker misses. + publish_after_commit(self._app, self._db) return await self._require_user(user_id) async def generate_reset_link(self, user_id: uuid.UUID, base_url: str) -> str: diff --git a/modules/users/users/admin/summary_cache.py b/modules/users/users/admin/summary_cache.py index c2f2478b6..a4ffc36a1 100644 --- a/modules/users/users/admin/summary_cache.py +++ b/modules/users/users/admin/summary_cache.py @@ -1,23 +1,26 @@ -"""Short-lived, app-local admin summaries, invalidated after user/role writes. - -Only writes that can change a summary invalidate it: a user created, deleted, -enabled/disabled, verified or invited; a role created, deleted or renamed; a -membership added or removed. Logins, failed-login counters, password changes -and session revocations touch ``users_user`` too but change no figure here, so -they publish nothing. A qualifying transaction publishes at most once, after -commit. Without a cross-process transport (or if a message is lost) another -worker shows old totals until :data:`TTL_SECONDS` expires; these figures are -informational and never gate access. +"""Short-lived, app-local admin summaries (state totals and role options). + +This worker forgets its entries when a transaction that wrote a ``User``, +``Role`` or ``UserRole`` through the ORM commits. Other workers learn of a +change only through :func:`publish_after_commit`, which the admin service's +write paths call explicitly; other writes (logins, lockout counters, password +changes) publish nothing. Without a cross-process transport, or if a message +is lost, another worker shows old totals until :data:`TTL_SECONDS` expires; +these figures are informational and never gate access. """ from __future__ import annotations import time -from typing import Literal, overload +from collections.abc import Awaitable, Callable +from typing import Any +from fastapi import FastAPI +from simple_module_core.invalidation import InvalidationBus from simple_module_db.callbacks import register_on_commit -from sqlalchemy import event, func, inspect, select -from sqlalchemy.ext.asyncio import async_session +from sqlalchemy import event, func, select +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import Session, SessionTransaction from sqlmodel import col from users.admin.user_state import STATUS_CONDITIONS @@ -26,23 +29,19 @@ CHANNEL = "users.admin_summary" TTL_SECONDS = 30.0 -# Columns whose change can move a summary figure; ``None`` means any change. -_COUNTED: dict[str, frozenset[str] | None] = { - User.__tablename__: frozenset({"is_active", "is_verified", "invited_at"}), - Role.__tablename__: frozenset({"name", "description"}), - UserRole.__tablename__: None, -} -_TOKEN = f"{CHANNEL}.publish" +_COUNTS, _ROLES = "counts", "roles" -async def load_counts(db) -> dict[str, int]: +async def load_counts(db: AsyncSession) -> dict[str, int]: + """One scan with ``COUNT(*) FILTER (WHERE …)`` per state rather than a + round-trip each (Postgres and SQLite >= 3.30 both support it).""" keys = ("active", "unverified", "invited") stmt = select(*(func.count().filter(STATUS_CONDITIONS[k]()) for k in keys)).select_from(User) row = (await db.execute(stmt)).one() return {key: int(value) for key, value in zip(keys, row, strict=True)} -async def load_roles(db) -> list[RoleListItem]: +async def load_roles(db: AsyncSession) -> list[RoleListItem]: stmt = ( select( Role.id, @@ -58,17 +57,17 @@ async def load_roles(db) -> list[RoleListItem]: class SummaryCache: - def __init__(self): - self.slots = {} + def __init__(self) -> None: + self.slots: dict[str, tuple[Any, float]] = {} self.generation = 0 self.wired = False - def forget(self): + def forget(self) -> None: self.slots.clear() self.generation += 1 -def _cache(app) -> SummaryCache: +def _cache(app: FastAPI) -> SummaryCache: """The app's cache, created on first use. Lazy so ``register_invalidations`` only subscribes: it must not assume an @@ -80,115 +79,82 @@ def _cache(app) -> SummaryCache: return cache -def _changes_summary(obj, dirty: bool) -> bool: - if not isinstance(obj, User | Role | UserRole): - return False - columns = _COUNTED[obj.__tablename__] - if not dirty or columns is None: - return True - attrs = inspect(obj).attrs - return any(attrs[name].history.has_changes() for name in columns) +async def _cached( + app: FastAPI | None, + db: AsyncSession, + key: str, + load: Callable[[AsyncSession], Awaitable[Any]], +) -> Any: + # Service-only callers and writes in progress need their transaction's own view. + if app is None or db.new or db.dirty or db.deleted or db.info.get(CHANNEL): + return await load(db) + cache = _cache(app) + slot = cache.slots.get(key) + now = time.monotonic() + if slot is not None and now < slot[1]: + return slot[0] + generation = cache.generation + # Never cache uncommitted rows or retain the request's transaction snapshot. + async with app.state.sm.db.session_factory() as committed_db: + value = await load(committed_db) + if generation == cache.generation: + cache.slots[key] = (value, now + TTL_SECONDS) + return value -def _statement_changes_summary(state) -> bool: - if not (state.is_insert or state.is_update or state.is_delete): - return False - name = getattr(getattr(state.statement, "table", None), "name", None) - if name not in _COUNTED: - return False - columns = _COUNTED[name] - if not state.is_update or columns is None: - return True - values = getattr(state.statement, "_values", None) - if not values: - # Bulk-by-primary-key or ordered values: unknown columns, stay safe. - return True - return any(getattr(key, "key", key) in columns for key in values) +async def get_role_summary(app: FastAPI | None, db: AsyncSession) -> list[RoleListItem]: + """Roles with their member counts, cached for :data:`TTL_SECONDS`.""" + value = await _cached(app, db, _ROLES, load_roles) + return [item.model_copy() for item in value] -@overload -async def get_summary(app, db, *, roles: Literal[True]) -> list[RoleListItem]: ... +async def get_state_counts(app: FastAPI | None, db: AsyncSession) -> dict[str, int]: + """Active/unverified/invited totals, cached for :data:`TTL_SECONDS`.""" + return dict(await _cached(app, db, _COUNTS, load_counts)) -@overload -async def get_summary(app, db, *, roles: Literal[False]) -> dict[str, int]: ... +def publish_after_commit(app: FastAPI | None, db: AsyncSession) -> None: + """Tell every worker to forget its summaries once ``db`` commits.""" + if app is None: + return + async def publish() -> None: + await app.state.sm.invalidation.publish(CHANNEL) -async def get_summary(app, db, *, roles): - load = load_roles if roles else load_counts - # Service-only callers and writes in progress need their transaction's own view. - if app is None or db.new or db.dirty or db.deleted or db.info.get(CHANNEL): - return await load(db) - cache = _cache(app) - slot = cache.slots.get(roles) - now = time.monotonic() - if slot is None or now >= slot[1]: - generation = cache.generation - # Never cache uncommitted rows or retain the request's transaction snapshot. - async with app.state.sm.db.session_factory() as committed_db: - value = await load(committed_db) - if generation == cache.generation: - cache.slots[roles] = (value, now + TTL_SECONDS) - else: - value = slot[0] - return [item.model_copy() for item in value] if roles else dict(value) - - -def subscribe(bus, app): + register_on_commit(db, publish) + + +def subscribe(bus: InvalidationBus, app: FastAPI) -> None: from users.roles_cache import CHANNEL as ROLES_CHANNEL bus.subscribe(CHANNEL, lambda _: _cache(app).forget()) bus.subscribe(ROLES_CHANNEL, lambda _: _cache(app).forget()) -def register_writes(app): +def register_writes(app: FastAPI) -> None: + """Forget this worker's summaries when a user/role write commits.""" cache = _cache(app) if cache.wired: return cache.wired = True - bus = app.state.sm.invalidation - - def written(session): - session.info[CHANNEL] = True - # One shared token per transaction: every qualifying flush queues a - # callback (so a savepoint rollback cannot drop the only one), but the - # first to run publishes and the rest are no-ops. - token = session.info.setdefault(_TOKEN, {"sent": False}) - - async def publish(): - if not token["sent"]: - token["sent"] = True - await bus.publish(CHANNEL) - - db = async_session(session) - assert db is not None - register_on_commit(db, publish) - - def before_flush(session, *_): - if any(_changes_summary(obj, False) for obj in (*session.new, *session.deleted)) or any( - _changes_summary(obj, True) for obj in session.dirty + + def before_flush(session: Session, *_: object) -> None: + if any( + isinstance(obj, User | Role | UserRole) + for obj in (*session.new, *session.dirty, *session.deleted) ): - written(session) - - def before_execute(state): - if _statement_changes_summary(state): - written(state.session) - - def after_commit(session): - if session.in_nested_transaction(): - return - session.info.pop(_TOKEN, None) - if session.info.pop(CHANNEL, False): - # Also covers fastapi-users adapters that commit before request finalization. + session.info[CHANNEL] = True + + def after_commit(session: Session) -> None: + # Also covers fastapi-users adapters that commit before request finalization. + if not session.in_nested_transaction() and session.info.pop(CHANNEL, False): cache.forget() - def after_rollback(session, transaction): + def after_rollback(session: Session, transaction: SessionTransaction) -> None: if transaction.parent is None: session.info.pop(CHANNEL, None) - session.info.pop(_TOKEN, None) session_class = app.state.sm.db.sync_session_class event.listen(session_class, "before_flush", before_flush) - event.listen(session_class, "do_orm_execute", before_execute) event.listen(session_class, "after_commit", after_commit) event.listen(session_class, "after_soft_rollback", after_rollback) diff --git a/modules/users/users/auth_local/token_api.py b/modules/users/users/auth_local/token_api.py index 9c7217b69..a23a37c1d 100644 --- a/modules/users/users/auth_local/token_api.py +++ b/modules/users/users/auth_local/token_api.py @@ -69,7 +69,6 @@ async def token_login( db: AsyncSession = Depends(get_db), ): """Exchange email + password for an access/refresh token pair.""" - helper = PASSWORD_HELPER # Same lockout as the session login (GH #530): a slot is taken before the # password is checked, and a success gives it back. attempt = login_attempt(request, body.email) @@ -93,13 +92,13 @@ async def token_login( # Constant-time: run bcrypt on a dummy hash to prevent timing-based # email enumeration (existing user + wrong password takes ~50ms for # bcrypt; missing user would be instant without this). - await asyncio.to_thread(helper.verify_and_update, body.password, _DUMMY_HASH) + await asyncio.to_thread(PASSWORD_HELPER.verify_and_update, body.password, _DUMMY_HASH) await rollback_session(db) locked = await record_failure(db, attempt, admission) raise locked or HTTPException(status_code=401, detail="Invalid credentials") verified, updated_hash = await asyncio.to_thread( - helper.verify_and_update, body.password, user.hashed_password + PASSWORD_HELPER.verify_and_update, body.password, user.hashed_password ) if not verified: await rollback_session(db) diff --git a/modules/users/users/core/__init__.py b/modules/users/users/core/__init__.py index f002e1469..38e57919f 100644 --- a/modules/users/users/core/__init__.py +++ b/modules/users/users/core/__init__.py @@ -1 +1 @@ -"""Users domain core: framework-free helpers.""" +"""Users domain core: password hashing boundaries.""" diff --git a/modules/users/users/core/password_hashing.py b/modules/users/users/core/password_hashing.py index 429251194..8c0f9c018 100644 --- a/modules/users/users/core/password_hashing.py +++ b/modules/users/users/core/password_hashing.py @@ -3,19 +3,29 @@ from __future__ import annotations import asyncio +from typing import Any import jwt +from fastapi import Request from fastapi_users import exceptions from fastapi_users.jwt import decode_jwt from fastapi_users.password import PasswordHelper +from users.contracts.schemas import UserCreate +from users.models import User + PASSWORD_HELPER = PasswordHelper() class PasswordHashingMixin: """Keep validation, database operations and lifecycle hooks on the event loop.""" - async def create(self, user_create, safe=False, request=None): + async def create( + self, + user_create: UserCreate, + safe: bool = False, + request: Request | None = None, + ) -> User: await self.validate_password(user_create.password, user_create) if await self.user_db.get_by_email(user_create.email) is not None: raise exceptions.UserAlreadyExists() @@ -28,7 +38,7 @@ async def create(self, user_create, safe=False, request=None): await self.on_after_register(user, request) return user - async def _update(self, user, update_dict): + async def _update(self, user: User, update_dict: dict[str, Any]) -> User: # Let upstream handle email uniqueness and every non-password field. data = dict(update_dict) if data.get("password") is not None: @@ -37,7 +47,9 @@ async def _update(self, user, update_dict): data["hashed_password"] = await asyncio.to_thread(self.password_helper.hash, password) return await super()._update(user, data) - async def reset_password(self, token, password, request=None): + async def reset_password( + self, token: str, password: str, request: Request | None = None + ) -> User: # Mirrors upstream token/fingerprint checks; only crypto leaves the loop. try: data = decode_jwt( From fd36d8023a7ba928e8d1b6c155bc28066e22537a Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Thu, 5 Nov 2026 19:33:20 +0100 Subject: [PATCH 4/5] =?UTF-8?q?fix:=20address=20code=20review=20findings?= =?UTF-8?q?=20(round=201,=20pass=201)=20=E2=80=94=20roleless=20create=20pu?= =?UTF-8?q?blishes;=20login=20bookkeeping=20keeps=20the=20summary=20cache?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claude-Session: https://claude.ai/code/session_01M9neheZZEe3sVpDi2S3zT4 --- CHANGELOG.md | 1 + docs/modules/users.md | 42 ++++++++++--------- .../tests/test_users_admin_summary_cache.py | 29 +++++++++++++ modules/users/users/admin/service.py | 11 +++-- modules/users/users/admin/summary_cache.py | 31 ++++++++++++-- 5 files changed, 88 insertions(+), 26 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3e454159f..8c3fd23db 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -42,6 +42,7 @@ All notable changes to this project are documented in this file. The format is b - `oidc` follows the module layout: `oidc.core.{constants,settings,presets,state,runtime}`, `oidc.identity.{provider,client,discovery,renewal,directory,persistence,tokens,jwks}` and `oidc.web.endpoints`. `oidc.{settings,provider,jwks}` still import, with a `DeprecationWarning`, until 0.0.46; other old `oidc.*` paths are gone. ### 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, diff --git a/docs/modules/users.md b/docs/modules/users.md index 1c166f11a..79529c65a 100644 --- a/docs/modules/users.md +++ b/docs/modules/users.md @@ -14,26 +14,6 @@ The default auth provider + user-management module: email/password login, OAuth/ ## Auth flow -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. - The module is built on [`fastapi-users`](https://fastapi-users.github.io/) for password hashing, registration, password reset, and verification. On top of that it layers: - A signed session cookie (`sm_auth` by default) — the auth module's `AuthMiddleware` calls `UsersAuthProvider.resolve_user`, which reads it on every request and populates `request.state.user`. Bearer tokens (`Authorization: Bearer `) resolve against `users_access_token`. @@ -263,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 diff --git a/modules/users/tests/test_users_admin_summary_cache.py b/modules/users/tests/test_users_admin_summary_cache.py index 741c9b191..c30b06451 100644 --- a/modules/users/tests/test_users_admin_summary_cache.py +++ b/modules/users/tests/test_users_admin_summary_cache.py @@ -233,3 +233,32 @@ def record(*args): 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 diff --git a/modules/users/users/admin/service.py b/modules/users/users/admin/service.py index 1d986a202..4237ea77a 100644 --- a/modules/users/users/admin/service.py +++ b/modules/users/users/admin/service.py @@ -11,7 +11,7 @@ from sqlmodel import col from users.admin.safety import _UserAdminSafety -from users.admin.summary_cache import publish_after_commit +from users.admin.summary_cache import publish_after_commit, publish_now from users.constants import ADMIN_ROLE_NAME from users.contracts.schemas import UserCreate from users.exceptions import ( @@ -59,10 +59,15 @@ async def create_user( is_superuser=False, ) user = await self._manager.create(user_create, safe=False) - # Queued after the adapter's own commit, so it waits for the roles too. - publish_after_commit(self._app, self._db) roles = await self._resolve_roles(role_names) + if roles: + # Queued after the adapter's own commit, so it waits for the roles. + publish_after_commit(self._app, self._db) + else: + # The adapter already committed and nothing else will be written, so + # the request would roll back and drop an on-commit callback. + await publish_now(self._app) for role in roles: self._db.add(UserRole(user_id=user.id, role_id=role.id, assigned_by=created_by)) if roles: diff --git a/modules/users/users/admin/summary_cache.py b/modules/users/users/admin/summary_cache.py index a4ffc36a1..078dc5507 100644 --- a/modules/users/users/admin/summary_cache.py +++ b/modules/users/users/admin/summary_cache.py @@ -11,6 +11,7 @@ from __future__ import annotations +import logging import time from collections.abc import Awaitable, Callable from typing import Any @@ -19,6 +20,7 @@ from simple_module_core.invalidation import InvalidationBus from simple_module_db.callbacks import register_on_commit from sqlalchemy import event, func, select +from sqlalchemy import inspect as sa_inspect from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Session, SessionTransaction from sqlmodel import col @@ -27,6 +29,8 @@ from users.contracts.schemas import RoleListItem from users.models import Role, User, UserRole +logger = logging.getLogger(__name__) + CHANNEL = "users.admin_summary" TTL_SECONDS = 30.0 _COUNTS, _ROLES = "counts", "roles" @@ -113,6 +117,16 @@ async def get_state_counts(app: FastAPI | None, db: AsyncSession) -> dict[str, i return dict(await _cached(app, db, _COUNTS, load_counts)) +async def publish_now(app: FastAPI | None) -> None: + """Tell every worker to forget its summaries; the write is already durable.""" + if app is None: + return + try: + await app.state.sm.invalidation.publish(CHANNEL) + except Exception: + logger.exception("users.admin_summary.publish_failed") + + def publish_after_commit(app: FastAPI | None, db: AsyncSession) -> None: """Tell every worker to forget its summaries once ``db`` commits.""" if app is None: @@ -131,6 +145,16 @@ def subscribe(bus: InvalidationBus, app: FastAPI) -> None: bus.subscribe(ROLES_CHANNEL, lambda _: _cache(app).forget()) +# The columns ``load_counts`` aggregates (see ``users.admin.user_state``); login +# bookkeeping (last_login_at, lockout counters, rehash, session_version) is not one. +_COUNTED_USER_COLUMNS = ("is_active", "is_verified", "invited_at") + + +def _counted_change(user: User) -> bool: + attrs = sa_inspect(user).attrs + return any(attrs[name].history.has_changes() for name in _COUNTED_USER_COLUMNS) + + def register_writes(app: FastAPI) -> None: """Forget this worker's summaries when a user/role write commits.""" cache = _cache(app) @@ -139,9 +163,10 @@ def register_writes(app: FastAPI) -> None: cache.wired = True def before_flush(session: Session, *_: object) -> None: - if any( - isinstance(obj, User | Role | UserRole) - for obj in (*session.new, *session.dirty, *session.deleted) + if ( + any(isinstance(obj, User | Role | UserRole) for obj in (*session.new, *session.deleted)) + or any(isinstance(obj, Role | UserRole) for obj in session.dirty) + or any(_counted_change(obj) for obj in session.dirty if isinstance(obj, User)) ): session.info[CHANNEL] = True From 651432f2b7cc0473c124222f1701620165907ea5 Mon Sep 17 00:00:00 2001 From: Anto Subash Date: Thu, 5 Nov 2026 19:35:00 +0100 Subject: [PATCH 5/5] test(users): mark the loop-lag heartbeat sleep as intentional Claude-Session: https://claude.ai/code/session_01M9neheZZEe3sVpDi2S3zT4 --- modules/users/tests/test_users_password_offload.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/modules/users/tests/test_users_password_offload.py b/modules/users/tests/test_users_password_offload.py index e39a00e06..1f3b46882 100644 --- a/modules/users/tests/test_users_password_offload.py +++ b/modules/users/tests/test_users_password_offload.py @@ -83,7 +83,7 @@ async def test_real_hashing_loop_lag(users_app, operation): async def heartbeat(): while running: started = time.perf_counter() - await asyncio.sleep(0.005) + await asyncio.sleep(0.005) # intentional-sleep: heartbeat measuring loop lag delays.append(max(0, time.perf_counter() - started - 0.005)) ticker = asyncio.create_task(heartbeat())