feat(BA-6626): app_config_fragment bulk repository layer & conditional-bulk primitives

Repository-layer foundation for app_config_fragment bulk operations (split out of BA-6618):

- Generic conditional-bulk primitives in repositories/base: ConditionalCreator / ConditionalUpdater
  / ConditionalPurger and BulkConditionalCreator / BulkConditionalUpdater / BulkConditionalPurger
  (a write spec paired with an only_if existence gate).
- WriteOps.bulk_conditional_create_partial / _update_partial / _purge_partial — each item runs in
  its own savepoint for partial success (gate-rejected / missing / failed items are reported, the
  rest proceed).
- Repository errors ConditionalWriteNotAllowed / ConditionalWriteTargetNotFound.
- app_config_fragment repository: AppConfigFragmentCreatorSpec as a plain CreatorSpec with
  scope-based rank; bulk_create / bulk_update / bulk_purge in db_source + repository; the
  AppConfigFragmentBulkWriteResult data type (succeeded + failed[index, message]).
- Repository and ops-provider unit tests.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: jopemachine

changes/12426.misc.md

@@ -0,0 +1 @@
+Add generic conditional-bulk repository primitives (`ConditionalCreator`/`Updater`/`Purger` paired with an `only_if` gate, and `WriteOps.bulk_conditional_*_partial` for per-item partial success) and the `app_config_fragment` bulk repository operations built on them.

src/ai/backend/manager/data/app_config_fragment/types.py

@@ -30,3 +30,23 @@ class AppConfigFragmentSearchResult:
     total_count: int
     has_next_page: bool
     has_previous_page: bool
+
+
+@dataclass(frozen=True)
+class AppConfigFragmentBulkItemError:
+    """One failed item of a partial bulk mutation: its batch position and a reason."""
+
+    index: int
+    message: str
+
+
+@dataclass(frozen=True)
+class AppConfigFragmentBulkWriteResult:
+    """Partial-success result of a bulk mutation.
+
+    ``succeeded`` are the fragments that were created/updated/purged; ``failed`` are the items
+    whose gate was rejected or whose write failed, each with its batch ``index`` and a reason.
+    """
+
+    succeeded: list[AppConfigFragmentData]
+    failed: list[AppConfigFragmentBulkItemError]

src/ai/backend/manager/errors/repository.py

@@ -15,6 +15,7 @@
     ErrorDomain,
     ErrorOperation,
 )
+from ai.backend.manager.errors.common import GenericForbidden, ObjectNotFound
 
 
 class RepositoryError(BackendAIError):
@@ -192,3 +193,25 @@ def error_code(self) -> ErrorCode:
             operation=ErrorOperation.GENERIC,
             error_detail=ErrorDetail.CONFLICT,
         )
+
+
+class ConditionalWriteNotAllowed(GenericForbidden):
+    """A gated bulk write item was rejected because its ``only_if`` gate did not pass.
+
+    Recorded as a per-item failure by the partial ``WriteOps.bulk_conditional_*_partial``
+    methods (the item is skipped, the rest proceed).
+    """
+
+    error_type = "https://api.backend.ai/probs/conditional-write-not-allowed"
+    error_title = "Conditional write is not allowed (gate not satisfied)."
+
+
+class ConditionalWriteTargetNotFound(ObjectNotFound):
+    """A gated bulk update/purge item targeted a row that does not exist.
+
+    Recorded as a per-item failure by the partial ``WriteOps.bulk_conditional_update_partial``
+    / ``bulk_conditional_purge_partial`` methods.
+    """
+
+    error_type = "https://api.backend.ai/probs/conditional-write-target-not-found"
+    object_name = "target row"

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

@@ -7,15 +7,26 @@
 
 from ai.backend.common.data.app_config.types import AppConfigScopeType
 from ai.backend.manager.models.app_config_fragment.row import AppConfigFragmentRow
-from ai.backend.manager.repositories.base.creator import DependentCreatorSpec
+from ai.backend.manager.repositories.base.creator import CreatorSpec
+
+# Merge precedence by scope: lower ranks merge first, so a more specific scope (higher rank)
+# overrides a broader one. Within one resolve there is at most one fragment per scope_type
+# (unique on ``(config_name, scope_type, scope_id)``), so these ranks are distinct per merge.
+_SCOPE_RANK: dict[AppConfigScopeType, int] = {
+    AppConfigScopeType.PUBLIC: 100,
+    AppConfigScopeType.DOMAIN: 200,
+    AppConfigScopeType.USER: 300,
+}
 
 
 @dataclass
-class AppConfigFragmentCreatorSpec(DependentCreatorSpec[int, AppConfigFragmentRow]):
-    """Fragment creator whose ``rank`` is assigned by the ops layer (next-value) at execution.
+class AppConfigFragmentCreatorSpec(CreatorSpec[AppConfigFragmentRow]):
+    """Fragment creator whose ``rank`` is derived from its ``scope_type`` (scope precedence).
 
-    ``build_row`` receives the computed next rank, so a newly created fragment is placed
-    after the existing fragments for the same ``config_name``.
+    A fragment's rank is fixed by its scope (``public`` < ``domain`` < ``user``), so a more
+    specific scope overrides a broader one when fragments are merged — independent of creation
+    order. Being a plain ``CreatorSpec`` (rank known at build time), it composes with
+    ``BulkConditionalCreator`` for gated bulk inserts.
     """
 
     config_name: str
@@ -24,11 +35,11 @@ class AppConfigFragmentCreatorSpec(DependentCreatorSpec[int, AppConfigFragmentRo
     config: dict[str, Any]
 
     @override
-    def build_row(self, next_rank: int) -> AppConfigFragmentRow:
+    def build_row(self) -> AppConfigFragmentRow:
         return AppConfigFragmentRow(
             config_name=self.config_name,
             scope_type=self.scope_type,
             scope_id=self.scope_id,
-            rank=next_rank,
+            rank=_SCOPE_RANK[self.scope_type],
             config=self.config,
         )

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

@@ -13,6 +13,8 @@
 from ai.backend.common.resilience.policies.retry import BackoffStrategy, RetryArgs, RetryPolicy
 from ai.backend.common.resilience.resilience import Resilience
 from ai.backend.manager.data.app_config_fragment.types import (
+    AppConfigFragmentBulkItemError,
+    AppConfigFragmentBulkWriteResult,
     AppConfigFragmentData,
     AppConfigFragmentSearchResult,
 )
@@ -21,27 +23,26 @@
     AppConfigFragmentWriteNotAllowed,
 )
 from ai.backend.manager.models.app_config_allow_list.row import AppConfigAllowListRow
-from ai.backend.manager.models.app_config_definition.row import AppConfigDefinitionRow
 from ai.backend.manager.models.app_config_fragment.row import AppConfigFragmentRow
 from ai.backend.manager.models.scopes import SearchScope
 from ai.backend.manager.repositories.app_config_fragment.creators import (
     AppConfigFragmentCreatorSpec,
 )
 from ai.backend.manager.repositories.base import (
     BatchQuerier,
+    BulkConditionalCreator,
+    BulkConditionalPurger,
+    BulkConditionalUpdater,
+    Creator,
     ExistsQuerier,
     Purger,
     Querier,
     Updater,
 )
-from ai.backend.manager.repositories.base.creator import NextValuePolicy
-from ai.backend.manager.repositories.ops import DBOpsProvider
+from ai.backend.manager.repositories.ops import DBOpsProvider, WriteOps
 
 __all__ = ("AppConfigFragmentDBSource",)
 
-# Gap between successive ranks, leaving room to re-order fragments without renumbering.
-RANK_GAP = 100
-
 app_config_fragment_db_source_resilience = Resilience(
     policies=[
         MetricPolicy(
@@ -70,29 +71,62 @@ class AppConfigFragmentDBSource:
     def __init__(self, ops_provider: DBOpsProvider) -> None:
         self._ops = ops_provider
 
+    async def _update_in_tx(
+        self,
+        w: WriteOps,
+        updater: Updater[AppConfigFragmentRow],
+        only_if: ExistsQuerier[AppConfigAllowListRow],
+    ) -> AppConfigFragmentData:
+        """Gate + update one fragment inside the caller's write transaction.
+
+        A missing fragment surfaces as the update returning None below.
+        """
+        if not await w.exists(only_if):
+            raise AppConfigFragmentWriteNotAllowed(
+                f"Writing app config fragment {updater.pk_value} is not allowed."
+            )
+        result = await w.update(updater)
+        if result is None:
+            raise AppConfigFragmentNotFound(f"App config fragment {updater.pk_value} not found")
+        return result.row.to_data()
+
+    async def _purge_in_tx(
+        self,
+        w: WriteOps,
+        purger: Purger[AppConfigFragmentRow],
+        only_if: ExistsQuerier[AppConfigAllowListRow],
+    ) -> AppConfigFragmentData:
+        """Gate + purge one fragment inside the caller's write transaction.
+
+        A missing fragment surfaces as the purge returning None below.
+        """
+        if not await w.exists(only_if):
+            raise AppConfigFragmentWriteNotAllowed(
+                f"Writing app config fragment {purger.pk_value} is not allowed."
+            )
+        result = await w.purge(purger)
+        if result is None:
+            raise AppConfigFragmentNotFound(f"App config fragment {purger.pk_value} not found")
+        return result.row.to_data()
+
     @app_config_fragment_db_source_resilience.apply()
     async def create(
         self,
         spec: AppConfigFragmentCreatorSpec,
         only_if: ExistsQuerier[AppConfigAllowListRow],
     ) -> AppConfigFragmentData:
-        policy = NextValuePolicy(
-            column=AppConfigFragmentRow.rank,
-            scope_condition=lambda: AppConfigFragmentRow.config_name == spec.config_name,
-            lock_selector=sa.select(AppConfigDefinitionRow).where(
-                AppConfigDefinitionRow.config_name == spec.config_name
-            ),
-            gap=RANK_GAP,
-        )
-        # ``only_if`` (built by the caller) and the write run in one transaction, so the gate
-        # check and the write commit atomically — no check-then-write race.
+        """Gate + create one fragment in a single write transaction.
+
+        The gate check and the write run in one transaction, so they commit atomically —
+        no check-then-write race. ``rank`` is derived from the fragment's ``scope_type``.
+        """
         async with self._ops.write_ops() as w:
             if not await w.exists(only_if):
                 raise AppConfigFragmentWriteNotAllowed(
                     f"Writing app config {spec.config_name!r} at scope "
                     f"{spec.scope_type.value!r} is not allowed."
                 )
-            created = await w.create_with_next_value(policy, spec)
+            created = await w.create(Creator(spec=spec))
             return created.row.to_data()
 
     @app_config_fragment_db_source_resilience.apply()
@@ -109,35 +143,78 @@ async def update(
         updater: Updater[AppConfigFragmentRow],
         only_if: ExistsQuerier[AppConfigAllowListRow],
     ) -> AppConfigFragmentData:
-        # Gate first, then write — both in one transaction so the check and the write commit
-        # atomically. A missing fragment surfaces as the update returning None below.
         async with self._ops.write_ops() as w:
-            if not await w.exists(only_if):
-                raise AppConfigFragmentWriteNotAllowed(
-                    f"Writing app config fragment {updater.pk_value} is not allowed."
-                )
-            result = await w.update(updater)
-            if result is None:
-                raise AppConfigFragmentNotFound(f"App config fragment {updater.pk_value} not found")
-            return result.row.to_data()
+            return await self._update_in_tx(w, updater, only_if)
 
     @app_config_fragment_db_source_resilience.apply()
     async def purge(
         self,
         purger: Purger[AppConfigFragmentRow],
         only_if: ExistsQuerier[AppConfigAllowListRow],
     ) -> AppConfigFragmentData:
-        # Gate first, then write — both in one transaction so the check and the write commit
-        # atomically. A missing fragment surfaces as the purge returning None below.
         async with self._ops.write_ops() as w:
-            if not await w.exists(only_if):
-                raise AppConfigFragmentWriteNotAllowed(
-                    f"Writing app config fragment {purger.pk_value} is not allowed."
-                )
-            result = await w.purge(purger)
-            if result is None:
-                raise AppConfigFragmentNotFound(f"App config fragment {purger.pk_value} not found")
-            return result.row.to_data()
+            return await self._purge_in_tx(w, purger, only_if)
+
+    @app_config_fragment_db_source_resilience.apply()
+    async def bulk_create(
+        self,
+        bulk: BulkConditionalCreator[AppConfigFragmentRow, AppConfigAllowListRow],
+    ) -> AppConfigFragmentBulkWriteResult:
+        """Create many fragments with partial success.
+
+        Each item is gated and inserted independently in its own savepoint: a rejected gate or a
+        failed insert is reported in ``failed`` (with its batch index) while the rest are
+        created. The whole batch shares one transaction, so the successful inserts commit together.
+        """
+        async with self._ops.write_ops() as w:
+            result = await w.bulk_conditional_create_partial(bulk)
+            return AppConfigFragmentBulkWriteResult(
+                succeeded=[row.to_data() for row in result.successes],
+                failed=[
+                    AppConfigFragmentBulkItemError(index=e.index, message=str(e.exception))
+                    for e in result.errors
+                ],
+            )
+
+    @app_config_fragment_db_source_resilience.apply()
+    async def bulk_update(
+        self,
+        bulk: BulkConditionalUpdater[AppConfigFragmentRow, AppConfigAllowListRow],
+    ) -> AppConfigFragmentBulkWriteResult:
+        """Update many fragments with partial success.
+
+        Each item is gated and updated independently in its own savepoint: a rejected gate, a
+        missing target, or a failed update is reported in ``failed`` while the rest are updated.
+        """
+        async with self._ops.write_ops() as w:
+            result = await w.bulk_conditional_update_partial(bulk)
+            return AppConfigFragmentBulkWriteResult(
+                succeeded=[row.to_data() for row in result.successes],
+                failed=[
+                    AppConfigFragmentBulkItemError(index=e.index, message=str(e.exception))
+                    for e in result.errors
+                ],
+            )
+
+    @app_config_fragment_db_source_resilience.apply()
+    async def bulk_purge(
+        self,
+        bulk: BulkConditionalPurger[AppConfigFragmentRow, AppConfigAllowListRow],
+    ) -> AppConfigFragmentBulkWriteResult:
+        """Purge many fragments with partial success.
+
+        Each item is gated and deleted independently in its own savepoint: a rejected gate, a
+        missing target, or a failed delete is reported in ``failed`` while the rest are purged.
+        """
+        async with self._ops.write_ops() as w:
+            result = await w.bulk_conditional_purge_partial(bulk)
+            return AppConfigFragmentBulkWriteResult(
+                succeeded=[row.to_data() for row in result.successes],
+                failed=[
+                    AppConfigFragmentBulkItemError(index=e.index, message=str(e.exception))
+                    for e in result.errors
+                ],
+            )
 
     @app_config_fragment_db_source_resilience.apply()
     async def admin_search(self, querier: BatchQuerier) -> AppConfigFragmentSearchResult:

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

@@ -9,6 +9,7 @@
 from ai.backend.common.resilience.policies.retry import BackoffStrategy, RetryArgs, RetryPolicy
 from ai.backend.common.resilience.resilience import Resilience
 from ai.backend.manager.data.app_config_fragment.types import (
+    AppConfigFragmentBulkWriteResult,
     AppConfigFragmentData,
     AppConfigFragmentSearchResult,
 )
@@ -21,7 +22,15 @@
 from ai.backend.manager.repositories.app_config_fragment.db_source import (
     AppConfigFragmentDBSource,
 )
-from ai.backend.manager.repositories.base import BatchQuerier, ExistsQuerier, Purger, Updater
+from ai.backend.manager.repositories.base import (
+    BatchQuerier,
+    BulkConditionalCreator,
+    BulkConditionalPurger,
+    BulkConditionalUpdater,
+    ExistsQuerier,
+    Purger,
+    Updater,
+)
 from ai.backend.manager.repositories.ops import DBOpsProvider
 
 __all__ = ("AppConfigFragmentRepository",)
@@ -91,3 +100,24 @@ async def scoped_search(
         self, querier: BatchQuerier, scopes: Sequence[SearchScope]
     ) -> AppConfigFragmentSearchResult:
         return await self._db_source.scoped_search(querier, scopes)
+
+    @app_config_fragment_repository_resilience.apply()
+    async def bulk_create(
+        self,
+        bulk: BulkConditionalCreator[AppConfigFragmentRow, AppConfigAllowListRow],
+    ) -> AppConfigFragmentBulkWriteResult:
+        return await self._db_source.bulk_create(bulk)
+
+    @app_config_fragment_repository_resilience.apply()
+    async def bulk_update(
+        self,
+        bulk: BulkConditionalUpdater[AppConfigFragmentRow, AppConfigAllowListRow],
+    ) -> AppConfigFragmentBulkWriteResult:
+        return await self._db_source.bulk_update(bulk)
+
+    @app_config_fragment_repository_resilience.apply()
+    async def bulk_purge(
+        self,
+        bulk: BulkConditionalPurger[AppConfigFragmentRow, AppConfigAllowListRow],
+    ) -> AppConfigFragmentBulkWriteResult:
+        return await self._db_source.bulk_purge(bulk)

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

@@ -1,4 +1,4 @@
-"""Types for app config fragment repository operations (search scopes)."""
+"""Types for app config fragment repository operations (search scopes, gated writes)."""
 
 from __future__ import annotations