Skip to content
Merged
Show file tree
Hide file tree
Changes from 8 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions openrag/core/ports/model_endpoint_repo.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"``."""
...
53 changes: 42 additions & 11 deletions openrag/services/orchestrators/model_endpoint_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -233,20 +233,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))
Comment thread
Ahmath-Gadji marked this conversation as resolved.
Resolved

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)
Comment thread
andyne13 marked this conversation as resolved.
Resolved

async def delete_model_endpoint(self, name: str, model_type: str) -> None:
Expand All @@ -255,26 +276,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
Expand Down
21 changes: 21 additions & 0 deletions openrag/services/orchestrators/preset_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand All @@ -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(
Expand Down
61 changes: 60 additions & 1 deletion openrag/services/persistence/model_endpoint_repo.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,15 @@
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):
Expand Down Expand Up @@ -132,5 +140,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.
Comment thread
andyne13 marked this conversation as resolved.
Resolved
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"]
34 changes: 28 additions & 6 deletions openrag/services/persistence/preset_repo.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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,
)
Comment on lines 97 to 106

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The rename looks much safer now, but I think there's still one small consistency edge case.
This transaction updates existing partition references from old_name to new_name, but partition preset references are stored as plain strings. If another request creates or updates a partition while the rename is in progress, it could still save old_name, leaving behind a stale reference.
Would it make sense to serialize preset renames with partition writes, for example by using a shared advisory lock or another locking mechanism?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Valid edge — a partition write committing old_name concurrently with the rename can still leave a stale reference. The robust fix (a shared advisory lock spanning preset rename AND the partition write path) is cross-cutting and touches the partition-write hot path, so we're deferring it to a focused follow-up after the pre-release rather than expanding this PR. Leaving this thread open to track it.

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)
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved

Expand Down
49 changes: 49 additions & 0 deletions tests/integration/api/test_model_endpoints.py
Original file line number Diff line number Diff line change
Expand Up @@ -79,3 +79,52 @@ def test_model_endpoint_crud_validate_and_default_selection(api_client):
api_client.post(f"/model-endpoints/llm/{original_default_llm}/set-default")
_delete_ignore_errors(api_client, f"/model-endpoints/llm/{endpoint_renamed}")
_delete_ignore_errors(api_client, f"/model-endpoints/llm/{endpoint_name}")


def test_update_is_default_keeps_single_default(api_client):
"""PUT {"is_default": true} via the update path must not create a second default.

Regression: ``update`` ran a bare ``SET is_default = true`` without clearing the
previous default, so a single PUT left two is_default=true rows for one type and
the 'default' alias resolved to whichever sorted last by name. The update path now
routes ``is_default`` through ``set_default`` (atomic clear-then-set).
"""
suffix = uuid.uuid4().hex[:8]
name_a = f"ci-llm-a-{suffix}"
name_b = f"ci-llm-b-{suffix}"
original_default_llm = None

try:
endpoints = api_client.get("/model-endpoints/", params={"model_type": "llm"})
_assert_success(endpoints, context="list llm endpoints")
original_default_llm = next(row["name"] for row in endpoints.json() if row["is_default"])

for name in (name_a, name_b):
created = api_client.post(
"/model-endpoints/",
json={
"name": name,
"model_type": "llm",
"endpoint": MOCK_VLLM_ENDPOINT,
"model_name": MOCK_CHAT_MODEL,
},
)
assert created.status_code == 201, created.text
assert created.json()["is_default"] is False

# Promote A through the dedicated endpoint, then promote B through the UPDATE
# path. The second promotion must demote A, not add a parallel default.
_assert_success(api_client.post(f"/model-endpoints/llm/{name_a}/set-default"), context="set-default A")

promote_b = api_client.put(f"/model-endpoints/llm/{name_b}", json={"is_default": True})
_assert_success(promote_b, context="update B is_default=true")
assert promote_b.json()["is_default"] is True

listing = api_client.get("/model-endpoints/", params={"model_type": "llm"}).json()
defaults = sorted(row["name"] for row in listing if row["is_default"])
assert defaults == [name_b], f"expected exactly one default ({name_b}), got {defaults}"
finally:
if original_default_llm is not None:
api_client.post(f"/model-endpoints/llm/{original_default_llm}/set-default")
_delete_ignore_errors(api_client, f"/model-endpoints/llm/{name_a}")
_delete_ignore_errors(api_client, f"/model-endpoints/llm/{name_b}")
Loading
Loading