fix(mcp): optimistic-lock concurrent MCP server PATCHes

src/backend/base/langflow/api/v2/mcp.py

@@ -6,6 +6,7 @@
 from fastapi import APIRouter, Body, Depends, HTTPException, UploadFile
 from lfx.base.agents.utils import safe_cache_get, safe_cache_set
 from lfx.base.mcp.util import update_tools
+from sqlalchemy import update
 from sqlalchemy.exc import IntegrityError
 from sqlmodel import select
 
@@ -28,6 +29,9 @@
 
 router = APIRouter(tags=["MCP"], prefix="/mcp")
 
+# Retry budget for the version-guarded merge-PATCH path (only same-server merges consume it).
+_MAX_UPSERT_RETRIES = 12
+
 
 def is_mcp_servers_locked(settings: object) -> bool:
     """Return True only when MCP lock is explicitly enabled.
@@ -386,86 +390,103 @@ async def update_server(
     server_config: dict,
     current_user: CurrentActiveUser,
     session: DbSession,
-    storage_service: Annotated[StorageService, Depends(get_storage_service)],
-    settings_service: Annotated[SettingsService, Depends(get_settings_service)],
+    storage_service: Annotated[StorageService, Depends(get_storage_service)],  # noqa: ARG001
+    settings_service: Annotated[SettingsService, Depends(get_settings_service)],  # noqa: ARG001
     *,
     check_existing: bool = False,
     delete: bool = False,
     merge_existing: bool = False,
 ):
     """Create, update, or delete one MCP server row for the user.
 
-    A single-row upsert on ``(user_id, name)`` replaces the file's non-atomic
-    read-modify-write: concurrent edits to *different* servers touch different rows
-    and never contend, so no update is lost at any worker/replica count - without an
-    in-process lock. A concurrent create of the *same* name is caught by the unique
-    constraint and folded into an update.
+    Upserts a single row keyed on ``(user_id, name)`` so concurrent edits to different
+    servers never contend. A merge PATCH guards its write with the row ``version`` and
+    retries on conflict, so two concurrent PATCHes to the same server merge instead of
+    last-writer-wins; a full replace updates unconditionally. ``current_user`` is read
+    once into ``user_id`` because a later commit/rollback can expire it and re-reading it
+    would attempt IO in an async context.
     """
-    result = await session.exec(
-        select(MCPServer).where(MCPServer.user_id == current_user.id, MCPServer.name == server_name)
-    )
-    existing = result.first()
+    user_id = current_user.id
 
     if delete:
+        result = await session.exec(
+            select(MCPServer).where(MCPServer.user_id == user_id, MCPServer.name == server_name)
+        )
+        existing = result.first()
         if existing is None:
-            raise HTTPException(status_code=500, detail="Server not found.")
+            raise HTTPException(status_code=404, detail="Server not found.")
         await session.delete(existing)
         await session.commit()
         _clear_server_cache(server_name)
         return None
 
-    if check_existing and existing is not None:
-        raise HTTPException(status_code=500, detail="Server already exists.")
-
-    if merge_existing and existing is not None:
-        # PATCH semantics: shallow-merge the new (partial) config over the existing
-        # one at the plaintext level, then re-encrypt (mirrors the file store's
-        # {**existing, **new}).
-        new_config = {**decrypt_mcp_config(existing.config or {}), **server_config}
-    else:
-        new_config = server_config
-
-    encrypted_config = encrypt_mcp_config(new_config)
-    transport = _derive_transport(new_config)
+    result = await session.exec(select(MCPServer).where(MCPServer.user_id == user_id, MCPServer.name == server_name))
+    existing = result.first()
 
-    if existing is None:
-        session.add(MCPServer(user_id=current_user.id, name=server_name, config=encrypted_config, transport=transport))
-    else:
-        existing.config = encrypted_config
-        existing.transport = transport
-        existing.version += 1
-        existing.updated_at = datetime.now(timezone.utc)
-        session.add(existing)
+    for _ in range(_MAX_UPSERT_RETRIES):
+        if check_existing and existing is not None:
+            raise HTTPException(status_code=409, detail="Server already exists.")
 
-    try:
-        await session.commit()
-    except IntegrityError:
-        # A concurrent request created the same (user, name) first (we came in with
-        # existing=None). Re-apply the create/merge rules against the winning row
-        # instead of overwriting it with our pre-race config.
-        await session.rollback()
-        result = await session.exec(
-            select(MCPServer).where(MCPServer.user_id == current_user.id, MCPServer.name == server_name)
-        )
-        existing = result.first()
         if existing is None:
-            raise
-        if check_existing:
-            raise HTTPException(status_code=500, detail="Server already exists.") from None
+            session.add(
+                MCPServer(
+                    user_id=user_id,
+                    name=server_name,
+                    config=encrypt_mcp_config(server_config),
+                    transport=_derive_transport(server_config),
+                )
+            )
+            try:
+                await session.commit()
+            except IntegrityError:
+                # Lost a create race; re-read the winner and fall through to the update path.
+                await session.rollback()
+                result = await session.exec(
+                    select(MCPServer).where(MCPServer.user_id == user_id, MCPServer.name == server_name)
+                )
+                existing = result.first()
+                continue
+            break
+
         if merge_existing:
             merged = {**decrypt_mcp_config(existing.config or {}), **server_config}
-            existing.config = encrypt_mcp_config(merged)
-            existing.transport = _derive_transport(merged)
-        else:
-            existing.config = encrypted_config
-            existing.transport = transport
+            updated = await session.execute(
+                update(MCPServer)
+                .where(MCPServer.id == existing.id, MCPServer.version == existing.version)
+                .values(
+                    config=encrypt_mcp_config(merged),
+                    transport=_derive_transport(merged),
+                    version=existing.version + 1,
+                    updated_at=datetime.now(timezone.utc),
+                )
+            )
+            await session.commit()
+            if updated.rowcount == 1:
+                break
+            # Version moved under us: expire and re-read so a concurrent delete reads as None.
+            session.expire(existing)
+            result = await session.exec(
+                select(MCPServer).where(MCPServer.user_id == user_id, MCPServer.name == server_name)
+            )
+            existing = result.first()
+            if existing is None:
+                raise HTTPException(status_code=404, detail="Server not found.")
+            continue
+
+        existing.config = encrypt_mcp_config(server_config)
+        existing.transport = _derive_transport(server_config)
         existing.version += 1
         existing.updated_at = datetime.now(timezone.utc)
         session.add(existing)
         await session.commit()
+        break
+    else:
+        raise HTTPException(status_code=409, detail="MCP server was updated concurrently; please retry.")
 
     _clear_server_cache(server_name)
-    return await get_server(server_name, current_user, session, storage_service, settings_service)
+    result = await session.exec(select(MCPServer).where(MCPServer.user_id == user_id, MCPServer.name == server_name))
+    row = result.first()
+    return decrypt_mcp_config(row.config or {}) if row is not None else None
 
 
 @router.post("/servers/{server_name}")

src/backend/tests/unit/api/v2/test_mcp_db_store.py

@@ -329,3 +329,64 @@ async def test_get_server_list_preserves_insertion_order():
 
     await engine.dispose()
     assert list(servers.keys()) == names, f"expected insertion order {names}, got {list(servers.keys())}"
+
+
+@pytest.mark.asyncio
+async def test_concurrent_merge_patches_preserve_all_fields(tmp_path):
+    """Concurrent merge PATCHes to the SAME server each set a distinct field; the version lock keeps all of them.
+
+    Without the version-guarded retry, all writers read the same base config and the last
+    commit wins, silently dropping the other fields (the gap flagged in review of #13976).
+    """
+    engine = await _file_engine(tmp_path / "mcp.db")
+    user = SimpleNamespace(id=uuid.uuid4())
+
+    with patch.multiple("langflow.api.v2.mcp", **CACHE_PATCH):
+        async with AsyncSession(engine, expire_on_commit=False) as session:
+            await update_server("svc", {"url": "https://x", "k0": "base"}, user, session, None, None)
+
+        async def patch_field(i: int):
+            async with AsyncSession(engine, expire_on_commit=False) as session:
+                await update_server("svc", {f"k{i}": str(i)}, user, session, None, None, merge_existing=True)
+
+        results = await asyncio.gather(*[patch_field(i) for i in range(1, 9)], return_exceptions=True)
+        async with AsyncSession(engine, expire_on_commit=False) as session:
+            final = await get_server("svc", user, session, None, None)
+
+    await engine.dispose()
+    errors = [r for r in results if isinstance(r, Exception)]
+    assert not errors, f"concurrent PATCHes raised: {errors}"
+    assert final["url"] == "https://x"
+    for i in range(1, 9):
+        assert final.get(f"k{i}") == str(i), f"lost concurrently-patched field k{i}; final={final}"
+
+
+@pytest.mark.asyncio
+async def test_concurrent_patch_and_delete_never_raises_raw_db_error(tmp_path):
+    """A merge PATCH racing a DELETE of the same server must fail cleanly (HTTPException), never a raw DB error."""
+    from fastapi import HTTPException
+
+    engine = await _file_engine(tmp_path / "mcp.db")
+    user = SimpleNamespace(id=uuid.uuid4())
+
+    async def create():
+        async with AsyncSession(engine, expire_on_commit=False) as session:
+            await update_server("svc", {"url": "https://x", "a": "1"}, user, session, None, None)
+
+    async def patch_it():
+        async with AsyncSession(engine, expire_on_commit=False) as session:
+            await update_server("svc", {"b": "2"}, user, session, None, None, merge_existing=True)
+
+    async def delete_it():
+        async with AsyncSession(engine, expire_on_commit=False) as session:
+            await update_server("svc", {}, user, session, None, None, delete=True)
+
+    with patch.multiple("langflow.api.v2.mcp", **CACHE_PATCH):
+        # Many rounds to make the "deleted between the PATCH's read and its guarded write" window likely.
+        for _ in range(15):
+            await create()
+            results = await asyncio.gather(patch_it(), delete_it(), return_exceptions=True)
+            raw = [r for r in results if isinstance(r, Exception) and not isinstance(r, HTTPException)]
+            assert not raw, f"PATCH/DELETE race raised a raw DB error: {[type(r).__name__ for r in raw]}"
+
+    await engine.dispose()