feat(BA-5837): add ValkeyCache source for AppConfigFragment merged-view reads

Re-introduces a Valkey-backed cache layer for the merged AppConfig
view, replacing the cache source that was dropped together with the
legacy `app_configs` table in #11265 (BA-5822). The hot path —
`AppConfigFragmentRepository.app_config(user_id, name)` — is the
WebUI bootstrap loop (BEP-1052 §6) and was previously a DB hit per
request.

Mirrors the legacy `AppConfigCacheSource` shape on top of the
`(user, name)`-keyed merged view:

- `cache_source/cache_source.py`:
  - `get_merged_config(user_id, name)` cache-aside read keyed by
    `app_config:merged:{user_id}:{name}`
  - `set_merged_config(merged, domain_name=...)` writes the deep-merged
    payload + indexes the cache key in `app_config:user_keys:{user_id}`
    and the user in `app_config:domain_users:{domain_name}` so
    invalidation runs without `SCAN`
  - `invalidate_for_scope(scope_type, scope_id)`:
    - `USER`: drop the user's per-name set
    - `DOMAIN` / `DOMAIN_USER_DEFAULTS`: cascade through every member
      of the domain user set
    - `PUBLIC`: rely on TTL (broad invalidation TBD)
  - `invalidate_for_user(user_id)` convenience for self-service writes
- `db_source.user_domain_name(user_id)` single-column lookup so the
  cache layer can tag merged-view entries with their owning domain
- `AppConfigFragmentRepository.app_config(...)` is now cache-aside
  (cache hit returns the cached payload + fresh DB fragments; cache
  miss writes through)
- `AppConfigFragmentAdminRepository.{create,update,purge}` invalidate
  the affected scope after the DB write
- `repositories.py` builds the cache source from
  `args.valkey_stat_client` and threads it into both repos
- All cache calls pass through `suppress_with_log` — cache failures
  log a warning and fall through to the DB

Author: jopemachine

changes/11297.feature.md

@@ -0,0 +1 @@
+Front the `AppConfigFragment` merged-view repository with a Valkey cache. Read path is cache-aside (`get_merged_config(user_id, name)` → DB fallback → cache write); admin / self-service bulk writes invalidate per-user or per-domain via membership indexes (no `SCAN`). Cache failures fall through to the DB (BEP-1052 §5).

src/ai/backend/manager/repositories/app_config_fragment/admin_repository.py

@@ -16,6 +16,9 @@
 )
 from ai.backend.manager.models.app_config_fragment.row import AppConfigFragmentRow
 from ai.backend.manager.models.utils import ExtendedAsyncSAEngine
+from ai.backend.manager.repositories.app_config_fragment.cache_source import (
+    AppConfigFragmentCacheSource,
+)
 from ai.backend.manager.repositories.app_config_fragment.creators import (
     AppConfigFragmentCreatorSpec,
 )
@@ -64,9 +67,15 @@ class AppConfigFragmentAdminRepository:
     """
 
     _db_source: AppConfigFragmentDBSource
+    _cache_source: AppConfigFragmentCacheSource | None
 
-    def __init__(self, db: ExtendedAsyncSAEngine) -> None:
+    def __init__(
+        self,
+        db: ExtendedAsyncSAEngine,
+        cache_source: AppConfigFragmentCacheSource | None = None,
+    ) -> None:
         self._db_source = AppConfigFragmentDBSource(db)
+        self._cache_source = cache_source
 
     # ── Mutations ─────────────────────────────────────────────────
 
@@ -84,7 +93,9 @@ async def create(
                 extra_config=extra_config,
             ),
         )
-        return await self._db_source.create(creator)
+        result = await self._db_source.create(creator)
+        await self._invalidate(key)
+        return result
 
     @app_config_fragment_admin_repository_resilience.apply()
     async def update(
@@ -95,11 +106,21 @@ async def update(
         """Update a fragment by natural key. Raises
         ``AppConfigFragmentNotFound`` when missing."""
         spec = AppConfigFragmentUpdaterSpec(extra_config=extra_config)
-        return await self._db_source.update(key, spec)
+        result = await self._db_source.update(key, spec)
+        await self._invalidate(key)
+        return result
 
     @app_config_fragment_admin_repository_resilience.apply()
     async def purge(self, key: AppConfigFragmentKey) -> bool:
-        return await self._db_source.purge(key)
+        result = await self._db_source.purge(key)
+        if result:
+            await self._invalidate(key)
+        return result
+
+    async def _invalidate(self, key: AppConfigFragmentKey) -> None:
+        if self._cache_source is None:
+            return
+        await self._cache_source.invalidate_for_scope(key.scope_type, key.scope_id)
 
     # ── Cross-scope reads ────────────────────────────────────────
 

src/ai/backend/manager/repositories/app_config_fragment/cache_source/__init__.py

@@ -0,0 +1,3 @@
+from .cache_source import AppConfigFragmentCacheSource
+
+__all__ = ("AppConfigFragmentCacheSource",)

src/ai/backend/manager/repositories/app_config_fragment/cache_source/cache_source.py

@@ -0,0 +1,203 @@
+"""Cache source for AppConfigFragment merged-view reads.
+
+The hot path is `AppConfigFragmentRepository.app_config(user_id, name)`,
+the per-`(user, name)` deep-merge of every contributing fragment in
+the policy's `scope_sources` chain (BEP-1052 §5). Each merged value
+gets cached under `app_config:merged:{user_id}:{name}` with a TTL,
+and is invalidated whenever a contributing fragment changes.
+
+Membership indexes let invalidation work without `SCAN`:
+
+- ``app_config:user_keys:{user_id}`` — set of `(user_id, name)` cache
+  keys this user currently has cached. Per-user invalidation pops the
+  set in one `SMEMBERS` + `DEL`.
+- ``app_config:domain_users:{domain_name}`` — set of `user_id`s
+  currently observed in this domain. Per-domain invalidation expands
+  to a per-user invalidation for each member.
+
+Cache failures never break a request — every public method is wrapped
+in `suppress_with_log` so any Valkey error falls through to the DB.
+"""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from collections.abc import Mapping
+from typing import Any, cast
+
+from glide import Batch, ExpirySet, ExpiryType
+
+from ai.backend.common.json import dump_json, load_json
+from ai.backend.logging.utils import BraceStyleAdapter
+from ai.backend.manager.clients.valkey_client.valkey_cache import ValkeyCache
+from ai.backend.manager.data.app_config.types import AppConfigData
+from ai.backend.manager.data.app_config_fragment.types import AppConfigScopeType
+from ai.backend.manager.repositories.utils import suppress_with_log
+
+log = BraceStyleAdapter(logging.getLogger(__spec__.name))
+
+
+class AppConfigFragmentCacheSource:
+    """Valkey-backed cache for the per-`(user, name)` merged AppConfig view."""
+
+    _valkey_cache: ValkeyCache
+    _cache_ttl: int
+
+    def __init__(self, valkey_cache: ValkeyCache, cache_ttl: int = 600) -> None:
+        """Args:
+        valkey_cache: Manager-side Valkey cache client.
+        cache_ttl: TTL in seconds (default: 10 minutes).
+        """
+        self._valkey_cache = valkey_cache
+        self._cache_ttl = cache_ttl
+
+    # ── Key derivation ─────────────────────────────────────────────
+
+    @staticmethod
+    def _merged_key(user_id: uuid.UUID | str, name: str) -> str:
+        return f"app_config:merged:{user_id}:{name}"
+
+    @staticmethod
+    def _user_keys_set(user_id: uuid.UUID | str) -> str:
+        return f"app_config:user_keys:{user_id}"
+
+    @staticmethod
+    def _domain_users_set(domain_name: str) -> str:
+        return f"app_config:domain_users:{domain_name}"
+
+    # ── Read ───────────────────────────────────────────────────────
+
+    async def get_merged_config(
+        self,
+        user_id: uuid.UUID,
+        name: str,
+    ) -> Mapping[str, Any] | None:
+        """Return the cached `config` payload, or `None` on miss / failure.
+
+        Only the merged `config` mapping is cached — `fragments` are
+        cheap to recompute when needed and would bloat the cache.
+        """
+        with suppress_with_log([Exception], "Failed to read merged config from cache"):
+            cache_key = self._merged_key(user_id, name)
+            async with self._valkey_cache.client() as conn:
+                cached_value = await conn.get(cache_key)
+            if cached_value:
+                log.debug("Cache hit for merged config: {} {}", user_id, name)
+                return cast(Mapping[str, Any] | None, load_json(cached_value))
+            log.debug("Cache miss for merged config: {} {}", user_id, name)
+        return None
+
+    # ── Write ──────────────────────────────────────────────────────
+
+    async def set_merged_config(
+        self,
+        merged: AppConfigData,
+        domain_name: str | None = None,
+    ) -> None:
+        """Cache the merged `config` payload + index it for invalidation.
+
+        Indexes the cache key in the user's key set; if `domain_name`
+        is supplied, also adds the user to the domain's user set so
+        domain-level invalidation can cascade.
+        """
+        if merged.config is None:
+            # Nothing useful to cache — leave as miss-on-next-read.
+            return
+        with suppress_with_log([Exception], "Failed to write merged config to cache"):
+            user_id = str(merged.user_id)
+            cache_key = self._merged_key(user_id, merged.name)
+
+            batch = Batch(is_atomic=False)
+            batch.set(
+                cache_key,
+                dump_json(merged.config),
+                expiry=ExpirySet(ExpiryType.SEC, self._cache_ttl),
+            )
+            user_keys = self._user_keys_set(user_id)
+            batch.sadd(user_keys, [cache_key])
+            batch.expire(user_keys, self._cache_ttl)
+            if domain_name is not None:
+                domain_users = self._domain_users_set(domain_name)
+                batch.sadd(domain_users, [user_id])
+                batch.expire(domain_users, self._cache_ttl)
+
+            async with self._valkey_cache.client() as conn:
+                await conn.exec(batch, raise_on_error=True)
+
+            log.trace(
+                "Cached merged config for user {} name {} (domain={})",
+                user_id,
+                merged.name,
+                domain_name,
+            )
+
+    # ── Invalidate ────────────────────────────────────────────────
+
+    async def invalidate_for_scope(
+        self,
+        scope_type: AppConfigScopeType,
+        scope_id: str,
+    ) -> None:
+        """Invalidate every cached merged view affected by a fragment write.
+
+        `scope_id` is the user UUID for `USER`, the domain name for
+        `DOMAIN` / `DOMAIN_USER_DEFAULTS`, or the literal `"public"`
+        for `PUBLIC`. `PUBLIC` invalidation is intentionally not
+        wired here — its blast radius is the whole cache; rely on TTL
+        for now (admin-only, low-frequency operation).
+        """
+        with suppress_with_log([Exception], "Failed to invalidate merged-config cache"):
+            match scope_type:
+                case AppConfigScopeType.USER:
+                    await self._invalidate_user(scope_id)
+                case AppConfigScopeType.DOMAIN | AppConfigScopeType.DOMAIN_USER_DEFAULTS:
+                    await self._invalidate_domain(scope_id)
+                case AppConfigScopeType.PUBLIC:
+                    log.debug(
+                        "PUBLIC-scope invalidation not wired; relying on TTL ({}s)",
+                        self._cache_ttl,
+                    )
+
+    async def invalidate_for_user(self, user_id: uuid.UUID | str) -> None:
+        """Drop every cached merged view owned by `user_id`.
+
+        Convenience wrapper used by self-service bulk writes that
+        always operate on the caller's `USER` row.
+        """
+        with suppress_with_log([Exception], "Failed to invalidate user merged-config cache"):
+            await self._invalidate_user(str(user_id))
+
+    async def _invalidate_user(self, user_id: str) -> None:
+        user_keys = self._user_keys_set(user_id)
+        async with self._valkey_cache.client() as conn:
+            cached_keys = await conn.smembers(user_keys)
+        if not cached_keys:
+            log.debug("No cached keys for user {}, skipping invalidation", user_id)
+            return
+        keys_to_delete: list[str | bytes] = list(cached_keys)
+        keys_to_delete.append(user_keys)
+        async with self._valkey_cache.client() as conn:
+            removed = await conn.delete(keys_to_delete)
+        log.debug("Invalidated {} merged-config keys for user {}", removed, user_id)
+
+    async def _invalidate_domain(self, domain_name: str) -> None:
+        domain_users = self._domain_users_set(domain_name)
+        async with self._valkey_cache.client() as conn:
+            user_ids = await conn.smembers(domain_users)
+        if not user_ids:
+            log.debug(
+                "No tracked users for domain {}, skipping cache invalidation",
+                domain_name,
+            )
+            return
+        for raw in user_ids:
+            user_id = raw.decode() if isinstance(raw, bytes) else str(raw)
+            await self._invalidate_user(user_id)
+        async with self._valkey_cache.client() as conn:
+            await conn.delete([domain_users])
+        log.debug(
+            "Invalidated merged-config caches for {} users in domain {}",
+            len(user_ids),
+            domain_name,
+        )

src/ai/backend/manager/repositories/app_config_fragment/db_source/db_source.py

@@ -107,6 +107,19 @@ async def get_by_id(self, id: uuid.UUID) -> AppConfigFragmentData | None:
             )
             return row.to_data() if row is not None else None
 
+    @app_config_fragment_db_source_resilience.apply()
+    async def user_domain_name(self, user_id: uuid.UUID) -> str | None:
+        """Single-column lookup of a user's `domain_name`.
+
+        Used by the cache layer to tag merged-view entries with their
+        owning domain so domain-scoped fragment writes can target a
+        bounded user set during invalidation.
+        """
+        async with self._db.begin_readonly_session() as db_sess:
+            return await db_sess.scalar(
+                sa.select(UserRow.domain_name).where(UserRow.uuid == user_id)
+            )
+
     @app_config_fragment_db_source_resilience.apply()
     async def create(self, creator: Creator[AppConfigFragmentRow]) -> AppConfigFragmentData:
         """Insert a new fragment via the shared Creator helper.

src/ai/backend/manager/repositories/app_config_fragment/repositories.py

@@ -1,9 +1,13 @@
 from dataclasses import dataclass
 from typing import Self
 
+from ai.backend.manager.clients.valkey_client.valkey_cache import ValkeyCache
 from ai.backend.manager.repositories.app_config_fragment.admin_repository import (
     AppConfigFragmentAdminRepository,
 )
+from ai.backend.manager.repositories.app_config_fragment.cache_source import (
+    AppConfigFragmentCacheSource,
+)
 from ai.backend.manager.repositories.app_config_fragment.repository import (
     AppConfigFragmentRepository,
 )
@@ -17,7 +21,12 @@ class AppConfigFragmentRepositories:
 
     @classmethod
     def create(cls, args: RepositoryArgs) -> Self:
+        # The merged-view read path is fronted by a Valkey cache; the
+        # repository falls through to the DB transparently if the
+        # cache layer fails.
+        valkey_cache = ValkeyCache(args.valkey_stat_client._client)
+        cache_source = AppConfigFragmentCacheSource(valkey_cache)
         return cls(
-            repository=AppConfigFragmentRepository(args.db),
-            admin_repository=AppConfigFragmentAdminRepository(args.db),
+            repository=AppConfigFragmentRepository(args.db, cache_source=cache_source),
+            admin_repository=AppConfigFragmentAdminRepository(args.db, cache_source=cache_source),
         )

src/ai/backend/manager/repositories/app_config_fragment/repository.py

@@ -14,6 +14,9 @@
     AppConfigFragmentSearchResult,
 )
 from ai.backend.manager.models.utils import ExtendedAsyncSAEngine
+from ai.backend.manager.repositories.app_config_fragment.cache_source import (
+    AppConfigFragmentCacheSource,
+)
 from ai.backend.manager.repositories.app_config_fragment.db_source import (
     AppConfigFragmentDBSource,
 )
@@ -47,14 +50,23 @@ class AppConfigFragmentRepository:
     """Read-side repository for AppConfigFragment.
 
     Scope-bound reads on raw fragments plus the per-user merged
-    `AppConfig` view. Mutations and admin cross-scope reads live on
-    `AppConfigFragmentAdminRepository`.
+    `AppConfig` view (BEP-1052 §5). Mutations and admin cross-scope
+    reads live on `AppConfigFragmentAdminRepository`. Retry + metric
+    policies are applied at the DB-source layer; the merged-view read
+    path is fronted by a Valkey cache so repeated WebUI bootstrap
+    queries don't hammer the DB.
     """
 
     _db_source: AppConfigFragmentDBSource
+    _cache_source: AppConfigFragmentCacheSource | None
 
-    def __init__(self, db: ExtendedAsyncSAEngine) -> None:
+    def __init__(
+        self,
+        db: ExtendedAsyncSAEngine,
+        cache_source: AppConfigFragmentCacheSource | None = None,
+    ) -> None:
         self._db_source = AppConfigFragmentDBSource(db)
+        self._cache_source = cache_source
 
     # ── Raw fragment reads ────────────────────────────────────────
 
@@ -82,7 +94,31 @@ async def app_config(
         user_id: uuid.UUID,
         config_name: str,
     ) -> AppConfigData:
-        return await self._db_source.get_user_app_config(user_id, config_name)
+        """Cache-aside read for the per-`(user, name)` merged view.
+
+        Returns a fresh `AppConfigData` whose `config` payload may
+        come from cache (fragments still resolve from the DB on
+        demand). Cache failures fall through transparently.
+        """
+        if self._cache_source is not None:
+            cached_config = await self._cache_source.get_merged_config(user_id, config_name)
+            if cached_config is not None:
+                # Re-fetch the fragment list from the DB; only the
+                # deep-merged `config` payload is cached. Keeps the
+                # response-shape contract unchanged.
+                merged = await self._db_source.get_user_app_config(user_id, config_name)
+                return AppConfigData(
+                    user_id=merged.user_id,
+                    name=merged.name,
+                    fragments=merged.fragments,
+                    config=cached_config,
+                )
+
+        merged = await self._db_source.get_user_app_config(user_id, config_name)
+        if self._cache_source is not None:
+            domain_name = await self._db_source.user_domain_name(user_id)
+            await self._cache_source.set_merged_config(merged, domain_name=domain_name)
+        return merged
 
     @app_config_fragment_repository_resilience.apply()
     async def search_app_configs(