fix(registry): preset/model-endpoint mutation integrity (rename, default, delete)

openrag/core/ports/model_endpoint_repo.py

@@ -30,3 +30,10 @@ async def delete(self, name: str, model_type: str) -> bool: ...
 
     @abstractmethod
     async def set_default(self, model_type: str, name: str) -> None: ...
+
+    @abstractmethod
+    async def delete_and_promote_default(self, name: str, model_type: str) -> tuple[str, str | None]:
+        """Atomically delete an endpoint and, if it was the default, promote a
+        survivor. Decides under a row lock. Returns ``(status, promoted_name)``
+        where status is ``"not_found" | "last" | "ok"``."""
+        ...

openrag/services/orchestrators/model_endpoint_service.py

@@ -209,6 +209,11 @@ async def create_model_endpoint(self, row: ModelEndpointRow) -> ModelEndpointRow
             )
         result = await self._repo.create(row)
         await self.load_all()
+        if row.is_default:
+            # The new endpoint became the default (repo demoted the previous one in
+            # the same transaction), so the cached 'default' alias client — built
+            # against the old default — is stale and must be evicted.
+            self._invalidate_client_cache(row.model_type, "default")
         return result
 
     async def get_model_endpoint(self, name: str, model_type: str) -> ModelEndpointRow:
@@ -233,20 +238,41 @@ async def update_model_endpoint(self, name: str, model_type: str, **fields: obje
             raise NotFoundError(f"Endpoint '{name}' of type '{model_type}' not found.")
 
         new_name: str | None = fields.pop("new_name", None)  # type: ignore[assignment]
+        # 'is_default' is not a plain column edit: flipping it must clear the previous
+        # default in the same transaction, so route a truthy value through set_default
+        # (atomic clear-then-set) instead of letting the repo write a second
+        # is_default=true row. A false/None value is a no-op here — you switch the
+        # default by promoting another endpoint, never by leaving the type with none.
+        promote_to_default = bool(fields.pop("is_default", None))
 
         if fields:
             updated = await self._repo.update(name, model_type, **fields)
         else:
             updated = existing
 
         effective_name = name
+        renamed_from: str | None = None
         if new_name and new_name != name:
             await self._repo.rename(name, model_type, new_name)
             effective_name = new_name
-            self._invalidate_client_cache(model_type, name)
-
-        self._invalidate_client_cache(model_type, effective_name)
+            renamed_from = name
+
+        if promote_to_default:
+            # Clears any prior default and sets this row in one transaction, then
+            # re-points the 'default' alias to it — never leaves two defaults.
+            await self._repo.set_default(model_type, effective_name)
+        # The 'default' alias client is stale when this endpoint becomes the default
+        # OR was already the default (its config just changed).
+        evict_default = bool(promote_to_default or existing.is_default)
+
+        # Reload the in-memory config FIRST, then evict, so a client rebuilt during
+        # the reload window from the old config cannot survive in the cache.
         await self.load_all()
+        if renamed_from is not None:
+            self._invalidate_client_cache(model_type, renamed_from)
+        self._invalidate_client_cache(model_type, effective_name)
+        if evict_default:
+            self._invalidate_client_cache(model_type, "default")
         return await self._repo.get(effective_name, model_type) or (updated or existing)
 
     async def delete_model_endpoint(self, name: str, model_type: str) -> None:
@@ -255,26 +281,36 @@ async def delete_model_endpoint(self, name: str, model_type: str) -> None:
         Raises 404 if not found, 422 if it is the last endpoint of its type
         (would leave components with no fallback).
         """
-        existing = await self._repo.get(name, model_type)
-        if existing is None:
+        # The last-endpoint guard and the survivor/default choice are made INSIDE
+        # the repo's locked transaction (not from a stale snapshot here), so
+        # concurrent deletes of the same type can't both pass the count check or
+        # promote an already-deleted survivor — which would leave the type with no
+        # endpoint / no default. The repo reports what happened.
+        status, promoted = await self._repo.delete_and_promote_default(name, model_type)
+        if status == "not_found":
             raise NotFoundError(f"Endpoint '{name}' of type '{model_type}' not found.")
-
-        all_of_type = await self._repo.list_all(model_type=model_type)
-        if len(all_of_type) <= 1:
+        if status == "last":
             raise ValidationError(f"Cannot delete the last '{model_type}' endpoint. Register a replacement first.")
 
-        await self._repo.delete(name, model_type)
-        self._invalidate_client_cache(model_type, name)
+        # Reload the in-memory config FIRST, then evict: evicting before load_all
+        # leaves a window where a concurrent request rebuilds a client from the old
+        # config and re-caches it; evicting after the reload drops any such stale
+        # client so the next request rebuilds from the fresh config.
         await self.load_all()
+        self._invalidate_client_cache(model_type, name)
+        if promoted is not None:
+            # The deleted endpoint was the default; its 'default' alias client is now stale.
+            self._invalidate_client_cache(model_type, "default")
 
     async def set_default(self, model_type: str, name: str) -> None:
         """Promote ``name`` to the default endpoint for ``model_type``."""
         existing = await self._repo.get(name, model_type)
         if existing is None:
             raise NotFoundError(f"Endpoint '{name}' of type '{model_type}' not found.")
         await self._repo.set_default(model_type, name)
-        self._invalidate_client_cache(model_type, "default")
+        # Reload before evicting so a client rebuilt during the window can't survive.
         await self.load_all()
+        self._invalidate_client_cache(model_type, "default")
 
     # ------------------------------------------------------------------
     # Endpoint validation

openrag/services/orchestrators/preset_service.py

@@ -27,6 +27,12 @@
 
 _VALID_PRESET_TYPES = frozenset({"indexation", "retrieval"})
 
+# The 'default' preset is the system fallback: partitions.{indexation,retrieval}_preset
+# default to this name (schema server_default + create_partition), and
+# resolve_partition_row raises if it is missing. It must always exist under this
+# name, so renaming it away or deleting it is rejected. Editing its config is fine.
+_RESERVED_PRESET_NAME = "default"
+
 _DEFAULT_SEEDS: dict[str, dict[str, dict[str, Any]]] = {
     # ``enable_contextualization`` is intentionally omitted from the ``default``
     # indexation preset below — it is injected at seed time from the global
@@ -210,6 +216,15 @@ async def update_preset(self, name: str, preset_type: str, **fields: object) ->
         effective_name = name
 
         if new_name and new_name != name:
+            if name == _RESERVED_PRESET_NAME:
+                raise ValidationError(
+                    f"Cannot rename the reserved '{_RESERVED_PRESET_NAME}' {preset_type} preset: "
+                    "it is the system fallback every partition resolves to."
+                )
+            if await self._repo.get(new_name, preset_type) is not None:
+                raise ValidationError(
+                    f"Cannot rename preset to '{new_name}': a {preset_type} preset with that name already exists."
+                )
             effective_name = new_name
             result = await self._repo.rename(name, effective_name, preset_type, effective_config)
         else:
@@ -230,6 +245,12 @@ async def delete_preset(self, name: str, preset_type: str) -> None:
         if existing is None:
             raise NotFoundError(f"Preset '{name}' of type '{preset_type}' not found.")
 
+        if name == _RESERVED_PRESET_NAME:
+            raise ValidationError(
+                f"Cannot delete the reserved '{_RESERVED_PRESET_NAME}' {preset_type} preset: "
+                "it is the system fallback every partition resolves to."
+            )
+
         count = await self._repo.count_partitions_using(name, preset_type)
         if count > 0:
             raise ValidationError(

openrag/services/persistence/model_endpoint_repo.py

@@ -12,11 +12,20 @@
 
 from core.config.model_endpoints import ModelEndpointRow
 from core.ports.model_endpoint_repo import ModelEndpointRepository
+from core.utils.exceptions import NotFoundError
 
 if TYPE_CHECKING:
     import asyncpg
 
-_ALLOWED_UPDATE_FIELDS = frozenset({"endpoint", "model_name", "batch_size", "timeout", "extra", "is_default"})
+# 'is_default' is deliberately excluded: a bare ``UPDATE ... SET is_default = true``
+# cannot clear the previous default in the same statement, so it would leave two
+# is_default=true rows for one model_type — and load_all() then resolves the
+# 'default' alias to whichever endpoint sorts last by name, not the one the caller
+# picked. (Verified against the live admin API: a single PUT of {"is_default": true}
+# on a non-default endpoint yielded two defaults for the type.) Promotion must go
+# through set_default / delete_and_promote_default, which clear-then-set inside one
+# transaction; ModelEndpointService.update_model_endpoint routes is_default there.
+_ALLOWED_UPDATE_FIELDS = frozenset({"endpoint", "model_name", "batch_size", "timeout", "extra"})
 
 
 class PgModelEndpointRepository(ModelEndpointRepository):
@@ -45,22 +54,34 @@ def _to_model(row: asyncpg.Record) -> ModelEndpointRow:
         )
 
     async def create(self, row: ModelEndpointRow) -> ModelEndpointRow:
-        rec = await self.pool.fetchrow(
-            """
-            INSERT INTO model_endpoints
-                (name, model_type, endpoint, model_name, batch_size, timeout, extra, is_default)
-            VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8)
-            RETURNING *
-            """,
-            row.name,
-            row.model_type,
-            row.endpoint,
-            row.model_name,
-            row.batch_size,
-            row.timeout,
-            row.extra,
-            row.is_default,
-        )
+        # A bare INSERT with is_default=true cannot clear the previous default, so
+        # POST /model-endpoints/ {"is_default": true} would leave two is_default=true
+        # rows for one model_type (same hazard the update path avoids by routing
+        # through set_default). Demote any existing default in the SAME transaction
+        # as the insert so the new endpoint becomes the sole default atomically.
+        async with self.pool.acquire() as conn:
+            async with conn.transaction():
+                if row.is_default:
+                    await conn.execute(
+                        "UPDATE model_endpoints SET is_default = false, updated_at = now() WHERE model_type = $1",
+                        row.model_type,
+                    )
+                rec = await conn.fetchrow(
+                    """
+                    INSERT INTO model_endpoints
+                        (name, model_type, endpoint, model_name, batch_size, timeout, extra, is_default)
+                    VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8)
+                    RETURNING *
+                    """,
+                    row.name,
+                    row.model_type,
+                    row.endpoint,
+                    row.model_name,
+                    row.batch_size,
+                    row.timeout,
+                    row.extra,
+                    row.is_default,
+                )
         return self._to_model(rec)
 
     async def get(self, name: str, model_type: str) -> ModelEndpointRow | None:
@@ -119,8 +140,25 @@ async def delete(self, name: str, model_type: str) -> bool:
         return result == "DELETE 1"
 
     async def set_default(self, model_type: str, name: str) -> None:
+        """Promote ``name`` to the default for ``model_type``, atomically.
+
+        Locks the type's rows (FOR UPDATE) and confirms ``name`` still exists
+        *inside* the transaction before clearing the old default. Without the
+        lock + existence check, a concurrent delete of ``name`` between the
+        caller's existence check and this transaction would make the second
+        UPDATE match 0 rows AFTER the first already cleared the previous default
+        — leaving the type with no default at all. Same invariant
+        ``delete_and_promote_default`` protects. Raises ``NotFoundError`` if the
+        target endpoint is gone.
+        """
         async with self.pool.acquire() as conn:
             async with conn.transaction():
+                rows = await conn.fetch(
+                    "SELECT name FROM model_endpoints WHERE model_type = $1 FOR UPDATE",
+                    model_type,
+                )
+                if name not in {r["name"] for r in rows}:
+                    raise NotFoundError(f"Endpoint '{name}' of type '{model_type}' not found.")
                 await conn.execute(
                     "UPDATE model_endpoints SET is_default = false, updated_at = now() WHERE model_type = $1",
                     model_type,
@@ -132,5 +170,56 @@ async def set_default(self, model_type: str, name: str) -> None:
                     model_type,
                 )
 
+    async def delete_and_promote_default(self, name: str, model_type: str) -> tuple[str, str | None]:
+        """Delete an endpoint and, if it was the default, promote a survivor to
+        default — all atomically and decided under a row lock.
+
+        Locking and deciding inside one transaction means concurrent deletes of the
+        same model type can't both pass a stale last-endpoint check or promote an
+        already-deleted survivor, so the type is never left with no endpoint or no
+        default. Returns ``(status, promoted_name)`` where ``status`` is
+        ``"not_found" | "last" | "ok"`` and ``promoted_name`` is set only when a
+        deleted default was replaced.
+        """
+        # Lock every row of this model_type (FOR UPDATE) so concurrent deletes of
+        # the same type serialize, then make the last-endpoint guard and survivor
+        # choice from the LOCKED, current state — not a stale snapshot. Otherwise two
+        # concurrent deletes could each pass a snapshot count check and remove the
+        # last row, or promote a survivor that another tx just deleted, leaving the
+        # type with no endpoint / no default. Returns (status, promoted_name):
+        # status is "not_found" | "last" | "ok"; promoted_name is set only when the
+        # deleted row was the default and a survivor was promoted.
+        async with self.pool.acquire() as conn:
+            async with conn.transaction():
+                rows = await conn.fetch(
+                    "SELECT name, is_default FROM model_endpoints WHERE model_type = $1 ORDER BY name FOR UPDATE",
+                    model_type,
+                )
+                names = [r["name"] for r in rows]
+                if name not in names:
+                    return ("not_found", None)
+                if len(names) <= 1:
+                    return ("last", None)
+                was_default = next(r["is_default"] for r in rows if r["name"] == name)
+                await conn.execute(
+                    "DELETE FROM model_endpoints WHERE name = $1 AND model_type = $2",
+                    name,
+                    model_type,
+                )
+                promoted: str | None = None
+                if was_default:
+                    promoted = next(n for n in names if n != name)  # deterministic: first by name
+                    await conn.execute(
+                        "UPDATE model_endpoints SET is_default = false, updated_at = now() WHERE model_type = $1",
+                        model_type,
+                    )
+                    await conn.execute(
+                        "UPDATE model_endpoints SET is_default = true, updated_at = now() "
+                        "WHERE name = $1 AND model_type = $2",
+                        promoted,
+                        model_type,
+                    )
+                return ("ok", promoted)
+
 
 __all__ = ["PgModelEndpointRepository"]

openrag/services/persistence/preset_repo.py

@@ -11,6 +11,7 @@
 from typing import TYPE_CHECKING
 
 from core.ports.preset_repo import PresetRepository
+from core.utils.exceptions import NotFoundError
 
 if TYPE_CHECKING:
     import asyncpg
@@ -79,18 +80,39 @@ async def upsert(self, name: str, preset_type: str, config: dict) -> dict:
         return self._row_to_dict(rec)
 
     async def rename(self, old_name: str, new_name: str, preset_type: str, config: dict) -> dict:
+        """Rename a preset to ``new_name`` (also updating its config) and repoint
+        every referencing partition, atomically.
+
+        Raises :class:`NotFoundError` if ``old_name`` no longer exists; a collision
+        with an existing ``new_name`` is rejected by the ``(name, preset_type)``
+        unique constraint. Returns the updated preset row as a dict.
+        """
+        # Rename in place with a plain UPDATE (not an upsert): the
+        # (name, preset_type) unique constraint then rejects a collision with an
+        # existing target name instead of an upsert silently overwriting it, and
+        # the row keeps its identity/created_at. Partitions reference a preset by
+        # its name string (partitions.{col}) with no FK, so referencing partitions
+        # are repointed in the same transaction or they would orphan.
+        col = _preset_column(preset_type)
         async with self.pool.acquire() as conn:
             async with conn.transaction():
-                await conn.execute(
-                    "DELETE FROM pipeline_presets WHERE name = $1 AND preset_type = $2",
+                rec = await conn.fetchrow(
+                    "UPDATE pipeline_presets SET name = $2, config = $3::jsonb, updated_at = now() "
+                    "WHERE name = $1 AND preset_type = $4 RETURNING *",
                     old_name,
+                    new_name,
+                    config,
                     preset_type,
                 )
-                rec = await conn.fetchrow(
-                    _UPSERT_PRESET_SQL,
+                if rec is None:
+                    # old_name vanished between the service's existence check and
+                    # this UPDATE (concurrent delete) — fail cleanly instead of
+                    # blowing up in _row_to_dict(None).
+                    raise NotFoundError(f"Preset '{old_name}' of type '{preset_type}' not found.")
+                await conn.execute(
+                    f"UPDATE partitions SET {col} = $1 WHERE {col} = $2",
                     new_name,
-                    preset_type,
-                    config,
+                    old_name,
                 )
         return self._row_to_dict(rec)