Tech-debt batch 13: apply role grants, module settings and flag overrides after commit, across workers (#621)

CHANGELOG.md

@@ -17,6 +17,9 @@ All notable changes to this project are documented in this file. The format is b
 ### Fixed
 - Database: discard `on_commit` callbacks registered inside a rolled-back savepoint,
   including released child savepoints, while preserving callbacks from surviving work (#631).
+- Role-permission edits, module settings updates/clears and system/tenant feature-flag
+  overrides now apply only after commit and reload across workers through the
+  invalidation bus. Periodic reloads bound staleness when messages are missed (#621).
 - Keycloak user-cache inserts and profile updates now commit in an independent transaction,
   preserving user IDs across logins and bearer requests. Concurrent first logins reuse the
   winning cache row; bearer requests throttle last-login writes to once per five minutes (#618).

docs/framework/invalidation.md

@@ -12,11 +12,13 @@ Several modules cache a row that is cheap to read and expensive to read *often*:
 |---|---|---|---|
 | `User.session_version` | nearly every authenticated request | 30s | `users.session_version_cache` |
 | aggregate storage size | the files index | 30s | `file_storage` |
-| hydrated settings objects | every request that reads a setting | **never** | `settings` |
+| hydrated settings objects | every request that reads a setting | 30s periodic reload | `settings` |
+| role permission overlays | every permission check | 30s periodic reload | `permissions` |
+| feature flag overrides | every flag check | 30s periodic reload | `feature_flags` |
 
 The first two share one failure shape: the worker that performed the write dropped its own entry immediately and **every other worker kept serving the stale value until its own entry expired**. Acceptable for "sign out everywhere"; much less so for a password change made because an account is believed compromised ([#318](https://github.com/antosubash/simple_module_python/issues/318)).
 
-The third is worse, and is worth stating separately because it is easy to assume it has a TTL too. `settings.reload.apply_changes_and_reload` swaps the settings object on `app.state.<module>` in the worker that handled the save and announces it on the **in-process** `EventBus`. There is no TTL and no re-read, so every other worker serves the old configuration **until it is restarted** — an admin editing a setting on a four-worker deployment changes it for one worker in four. That needs a re-read rather than a `pop`, which is why it is not wired here (see *Still per-process*).
+Role overlays, module settings and feature-flag overrides now reload from the committed DB on invalidation (#621). The writing worker publishes only from `db.on_commit`; rollback changes no live state. Reloads replace snapshots so deleted overrides disappear too. Settings reloads emit a local `SettingsReloaded` event when values change. A serialized periodic reload every 30 seconds provides the fallback when a message is lost or broadcasting is disabled; failed reads keep the last snapshot and retry on the next interval.
 
 The obvious fix — publish on Redis pub/sub — was not reachable from any of those modules. Redis belongs to the `background_tasks` plugin, so `users` would have to open a second connection with its own settings that can disagree with it; `SM009` makes a `framework/*` → plugin import an error, so the framework cannot borrow the plugin's client either. Hence the split below.
 
@@ -107,4 +109,4 @@ A standalone worker never builds the FastAPI app, so it installs no transport: i
 
 ## Still per-process
 
-Only `users.session_version` is wired so far. `file_storage`'s aggregate cache and `settings`' hot-swap have the same staleness shape and the same fix available; they are separate changes because each needs its own decision about what "forget" means for it — a settings hot-swap in particular has to *re-read*, not just drop, so it needs more than a `pop`.
+`file_storage`'s aggregate cache remains per-process. Role overlays (`permissions.role_grants`), module settings (`settings.modules`, keyed by package) and feature flags (`feature_flags.overrides`, both scopes) subscribe through their modules' `register_invalidations` hooks. Their refresh tasks start and stop with module lifespan.

framework/core/simple_module_core/feature_flags.py

@@ -71,6 +71,13 @@ def clear_override(self, name: str, tenant_id: str | None = None) -> None:
         else:
             self._tenant_overrides.pop((name, tenant_id), None)
 
+    def replace_overrides(
+        self, system: dict[str, bool], tenants: dict[tuple[str, str], bool]
+    ) -> None:
+        """Replace a complete DB snapshot, removing overrides absent from it."""
+        self._system_overrides = dict(system)
+        self._tenant_overrides = dict(tenants)
+
     def tenant_override(self, name: str, tenant_id: str) -> bool | None:
         """Return the per-tenant override for ``(name, tenant_id)`` if set, else None."""
         return self._tenant_overrides.get((name, tenant_id))

framework/core/simple_module_core/invalidation.py

@@ -9,11 +9,9 @@
 immediately, and **every other worker keeps serving the stale value until its own
 entry expires**.
 
-``settings`` has the same shape without the TTL — ``apply_changes_and_reload``
-hot-swaps the settings object in the worker that handled the save and announces it
-on the in-process :class:`~simple_module_core.events.EventBus`, so the others serve
-the old configuration until restart. It needs a re-read rather than an eviction, so
-it is a consumer this bus makes *possible* rather than one it already serves.
+``settings``, role overlays and feature-flag overrides had it too (#621); they now
+re-read the committed DB state on invalidation (and every 30 s as a fallback, see
+:mod:`simple_module_core.invalidation_refresh`) rather than evicting an entry.
 
 The obvious fix — publish the write on Redis pub/sub so every worker drops its
 entry at once — was not reachable from any of those modules. Redis belongs to

framework/core/simple_module_core/invalidation_refresh.py

@@ -0,0 +1,51 @@
+"""Reload DB-backed process state on invalidation and periodically after missed messages."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from collections.abc import Awaitable, Callable
+from contextlib import suppress
+
+from simple_module_core.invalidation import Invalidation, InvalidationBus
+
+logger = logging.getLogger(__name__)
+REFRESH_SECONDS = 30
+
+
+class InvalidationRefresh:
+    """Serialize reloads and own the bounded-staleness fallback's lifecycle."""
+
+    def __init__(
+        self, bus: InvalidationBus, channel: str, reload: Callable[[str | None], Awaitable[None]]
+    ) -> None:
+        self._reload = reload
+        self._lock = asyncio.Lock()
+        self._task: asyncio.Task | None = None
+        bus.subscribe(channel, self._receive)
+
+    async def _receive(self, message: Invalidation) -> None:
+        await self.refresh(message.key)
+
+    async def refresh(self, key: str | None = None) -> None:
+        async with self._lock:
+            await self._reload(key)
+
+    def start(self) -> None:
+        if self._task is None:
+            self._task = asyncio.create_task(self._periodic())
+
+    async def stop(self) -> None:
+        if self._task is not None:
+            self._task.cancel()
+            with suppress(asyncio.CancelledError):
+                await self._task
+            self._task = None
+
+    async def _periodic(self) -> None:
+        while True:
+            await asyncio.sleep(REFRESH_SECONDS)
+            try:
+                await self.refresh()
+            except Exception:
+                logger.exception("Periodic state reload failed; retrying on the next interval")

framework/core/simple_module_core/permissions.py

@@ -208,6 +208,14 @@ def set_role_overlay(self, role: str, permissions: Collection[str]) -> None:
         self._role_overlay[role] = set(permissions)
         self._invalidate()
 
+    def replace_role_overlays(self, overlays: dict[str, list[str]]) -> None:
+        """Replace the complete persisted snapshot, including removed grants."""
+        snapshot = {role: set(keys) for role, keys in overlays.items()}
+        if snapshot == self._role_overlay:  # periodic reloads: keep derived caches warm
+            return
+        self._role_overlay = snapshot
+        self._invalidate()
+
     @property
     def role_map(self) -> dict[str, list[str]]:
         """Merged role→permission mapping (``DEFAULT_ROLE_PERMISSIONS`` + module maps)."""

framework/core/tests/test_core_invalidation_refresh.py

@@ -0,0 +1,54 @@
+"""Missed messages and transient reload failures must not leave state stale forever."""
+
+import asyncio
+
+from simple_module_core.invalidation import InvalidationBus
+from simple_module_core.invalidation_refresh import InvalidationRefresh
+
+
+async def test_periodic_refresh_retries_without_a_message_and_stops(monkeypatch):
+    monkeypatch.setattr("simple_module_core.invalidation_refresh.REFRESH_SECONDS", 0.001)
+    attempts = []
+    refreshed = asyncio.Event()
+
+    async def reload(key):
+        attempts.append(key)
+        if len(attempts) == 1:
+            raise RuntimeError("transient DB failure")
+        refreshed.set()
+
+    refresh = InvalidationRefresh(InvalidationBus(), "test.snapshot", reload)
+    refresh.start()
+    try:
+        await asyncio.wait_for(refreshed.wait(), timeout=1)
+    finally:
+        await refresh.stop()
+    assert attempts[:2] == [None, None]
+    count = len(attempts)
+    await asyncio.sleep(0.01)  # intentional-sleep: proves no reload runs after stop()
+    assert len(attempts) == count
+
+
+async def test_reload_messages_are_serialized():
+    bus = InvalidationBus()
+    entered = asyncio.Event()
+    release = asyncio.Event()
+    keys = []
+
+    async def reload(key):
+        keys.append(key)
+        if key == "first":
+            entered.set()
+            await release.wait()
+
+    InvalidationRefresh(bus, "test.snapshot", reload)
+    first = asyncio.create_task(bus.publish("test.snapshot", key="first"))
+    await asyncio.wait_for(entered.wait(), timeout=1)
+    second = asyncio.create_task(bus.publish("test.snapshot", key="second"))
+    try:
+        await asyncio.sleep(0)
+        assert keys == ["first"]
+    finally:
+        release.set()
+        await asyncio.wait_for(asyncio.gather(first, second), timeout=1)
+    assert keys == ["first", "second"]

framework/core/tests/test_feature_flags.py

@@ -194,3 +194,16 @@ async def test_is_flag_enabled_reads_tenant_from_request_state(self):
 
         request.state.tenant_id = "other"
         assert is_flag_enabled(request, "beta_ui") is False
+
+
+async def test_reloading_overrides_removes_deleted_values_in_both_scopes():
+    registry = FeatureFlagRegistry()
+    registry.add(FeatureFlagDefinition("switch", default_enabled=True))
+    registry.set_override("switch", False)
+    registry.set_override("switch", False, tenant_id="acme")
+    registry.replace_overrides({}, {("switch", "globex"): False})
+    assert registry.is_enabled("switch") is True
+    assert registry.is_enabled("switch", "acme") is True
+    assert registry.is_enabled("switch", "globex") is False
+    registry.replace_overrides({}, {})
+    assert registry.is_enabled("switch", "globex") is True

framework/core/tests/test_permissions.py

@@ -141,3 +141,13 @@ async def test_it_reads_any_collection(self):
         """Callers hand it whatever they have — the middleware caches a set,
         a test hands a list."""
         assert grants(["users.manage"], "users.manage") is True
+
+
+async def test_reloading_role_overlays_removes_deleted_grants_but_keeps_code_grants():
+    registry = PermissionRegistry()
+    registry.map_role("editor", ["code.view"])
+    registry.set_role_overlay("editor", ["db.view"])
+    registry.replace_role_overlays({"reviewer": ["db.review"]})
+    assert registry.role_map["editor"] == ["code.view"]
+    assert registry.role_map["reviewer"] == ["db.review"]
+    assert registry.role_map["admin"] == ["*"]

framework/testing/simple_module_test/_schema.py

@@ -29,19 +29,33 @@ def _ensure_models_imported() -> list:
     return list(all_module_bases)
 
 
-@lru_cache(maxsize=1)
+_heads_cache: tuple[str, ...] | None = None
+
+
 def _alembic_heads() -> tuple[str, ...]:
-    """Cached head revisions — cannot change within a pytest run.
+    """Head revisions — cannot change within a pytest run, once found.
+
+    Only a non-empty result is cached. ``resolve_head_revisions`` returns ``()``
+    when ``alembic.ini`` is not found, which is what happens under a test that
+    points ``SM_PROJECT_ROOT`` at a tmp dir; caching that would leave every
+    later test in the worker unstamped while its app, resolving the real ini,
+    demands the heads ("Database is N revision(s) behind").
 
     Plural: each module's first migration sets its own ``branch_labels``, so
     the history has one head per module and a real upgraded database carries
     an ``alembic_version`` row for each. Stamping only one leaves the rest
     looking un-applied, which now reads as a behind-head schema and puts every
     test app behind the setup gate.
     """
+    global _heads_cache
     from simple_module_hosting.migrations import resolve_head_revisions
 
-    return resolve_head_revisions()
+    if _heads_cache is None:
+        heads = resolve_head_revisions()
+        if not heads:
+            return heads
+        _heads_cache = heads
+    return _heads_cache
 
 
 async def _create_all_tables(engine) -> None:

modules/branding/branding/service.py

@@ -8,7 +8,7 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, cast
 
 from branding.constants import FAVICON_URL, LOGO_DARK_URL, LOGO_URL, PACKAGE, TENANT_IMAGE_FIELDS
 from branding.contracts.schemas import BrandingOut
@@ -73,8 +73,11 @@ async def apply(self, changes: dict[str, Any]) -> BrandingOut:
         invalidation = getattr(self.app.state.sm, "invalidation", None)
         store = SettingsStore(SettingService(self.db, invalidation=invalidation))
         bus = self.app.state.sm.event_bus
-        await apply_changes_and_reload(self.app, bus, store, package=PACKAGE, changes=changes)
-        return self.current()
+        validated = await apply_changes_and_reload(
+            self.app, bus, store, package=PACKAGE, changes=changes
+        )
+        # Registered for PACKAGE, so the validated object is BrandingSettings.
+        return to_out(cast("BrandingSettings", validated))
 
     async def _swap_asset(self, field: str, file_id: str) -> BrandingOut:
         """Point *field* at *file_id* ("" to clear) and reap what it replaced.
@@ -86,9 +89,22 @@ async def _swap_asset(self, field: str, file_id: str) -> BrandingOut:
         file that is gone. A file another system image field still references
         (logo and dark logo sharing one upload) is kept.
         """
-        previous = getattr(self.app.state.branding.settings, field, "")
+        # The pre-save value comes from the DB-backed settings, not app.state:
+        # the in-process object only changes after commit.
+        from settings.service import SettingService
+        from settings.store import SettingsStore
+        from simple_module_core.settings_hydrate import hydrate_settings
+
+        from branding.settings import BrandingSettings
+
+        before = await hydrate_settings(
+            BrandingSettings, SettingsStore(SettingService(self.db)), PACKAGE, lenient=True
+        )
+        previous = getattr(before, field, "")
         out = await self.apply({field: file_id})
-        live = self.app.state.branding.settings
+        live = await hydrate_settings(
+            BrandingSettings, SettingsStore(SettingService(self.db)), PACKAGE, lenient=True
+        )
         still_used = {getattr(live, name, "") for name in TENANT_IMAGE_FIELDS}
         # No storage handed in (a hand-built service): nothing is reaped.
         replaced = previous and previous != file_id and previous not in still_used

modules/branding/tests/test_presets.py

@@ -11,6 +11,9 @@
 import httpx
 import pytest
 from branding.presets import BUILTIN_PRESETS, PRESET_FIELDS, BrandingPreset, find_preset
+from settings.invalidation import reload_package
+from settings.service import SettingService
+from settings.store import SettingsStore
 
 # ── Unit: definitions ──────────────────────────────────────────────────
 
@@ -104,7 +107,15 @@ async def test_applying_a_preset_preserves_identity_and_banner(
 async def test_applying_a_preset_keeps_an_uploaded_logo(
     app, authenticated_client: httpx.AsyncClient
 ) -> None:
-    app.state.branding.settings.logo_file_id = "11111111-1111-1111-1111-111111111111"
+    # Persist the logo: settings now reload from the DB after commit, so an
+    # in-memory-only assignment would (correctly) be discarded.
+    async with app.state.sm.db.session_factory() as db:
+        store = SettingsStore(SettingService(db))
+        await store.set_override(
+            "branding", "logo_file_id", "11111111-1111-1111-1111-111111111111", "string"
+        )
+        await db.commit()
+        await reload_package(app, app.state.sm.event_bus, store, "branding")
 
     await authenticated_client.post(f"/api/branding/presets/{BUILTIN_PRESETS[0].key}")
 

modules/feature_flags/feature_flags/deps.py

@@ -14,9 +14,10 @@
 
 
 async def get_feature_flag_service(
+    request: Request,
     db: AsyncSession = Depends(get_db),
 ) -> FeatureFlagService:
-    return FeatureFlagService(db)
+    return FeatureFlagService(db, app=request.app)
 
 
 async def get_feature_flag_registry(request: Request) -> FeatureFlagRegistry:

modules/feature_flags/feature_flags/endpoints/api.py

@@ -70,6 +70,8 @@ async def set_override(
     await service.set_override(name, body.enabled, registry=registry)
     view = service.build_view(registry, name)
     assert view is not None  # _ensure_registered already proved it exists
+    view.enabled = body.enabled
+    view.overridden = True
     return view
 
 
@@ -125,6 +127,8 @@ async def set_tenant_override(
     )
     view = service.build_view(registry, name, tenant_id=tenant_id)
     assert view is not None  # _ensure_registered already proved it exists
+    view.enabled = body.enabled
+    view.overridden = True
     return view
 
 

modules/feature_flags/feature_flags/invalidation.py

@@ -0,0 +1,17 @@
+"""Re-read flag overrides after commit or a remote invalidation."""
+
+from fastapi import FastAPI
+from simple_module_core.invalidation import InvalidationBus
+from simple_module_core.invalidation_refresh import InvalidationRefresh
+
+from feature_flags.service import FeatureFlagService
+
+CHANNEL = "feature_flags.overrides"
+
+
+def subscribe(bus: InvalidationBus, app: FastAPI) -> InvalidationRefresh:
+    async def reload(_key: str | None) -> None:
+        async with app.state.sm.db.session_factory() as db:
+            await FeatureFlagService(db).hydrate_registry(app.state.sm.feature_flags)
+
+    return InvalidationRefresh(bus, CHANNEL, reload)