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.
@@ -406,7 +410,7 @@ async def update_server(
     server_config: dict,
     current_user: CurrentActiveUser,
     session: DbSession,
-    storage_service: Annotated[StorageService, Depends(get_storage_service)],
+    storage_service: Annotated[StorageService, Depends(get_storage_service)],  # noqa: ARG001
     settings_service: Annotated[SettingsService, Depends(get_settings_service)],
     *,
     check_existing: bool = False,
@@ -415,83 +419,121 @@ async def update_server(
 ):
     """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.
     """
+    user_id = current_user.id
     settings = getattr(settings_service, "settings", None)
     if not delete:
         ensure_mcp_stdio_access(server_config, current_user, settings)
 
-    result = await session.exec(
-        select(MCPServer).where(MCPServer.user_id == current_user.id, MCPServer.name == server_name)
-    )
-    existing = result.first()
-
     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
-
-    ensure_mcp_stdio_access(new_config, current_user, settings)
-    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:
+                # The expected IntegrityError here is the duplicate-name race: re-read the
+                # winner and fall through to the update path. Any other integrity failure
+                # (e.g. a bad FK) leaves no winning row, so re-raise it instead of masking
+                # it as retries that end in a misleading 409.
+                await session.rollback()
+                result = await session.exec(
+                    select(MCPServer).where(MCPServer.user_id == user_id, MCPServer.name == server_name)
+                )
+                existing = result.first()
+                if existing is None:
+                    raise
+                continue
+            break
+
         if merge_existing:
             merged = {**decrypt_mcp_config(existing.config or {}), **server_config}
             ensure_mcp_stdio_access(merged, current_user, settings)
-            existing.config = encrypt_mcp_config(merged)
-            existing.transport = _derive_transport(merged)
-        else:
-            existing.config = encrypted_config
-            existing.transport = transport
-        existing.version += 1
-        existing.updated_at = datetime.now(timezone.utc)
-        session.add(existing)
+            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
+
+        # Full replace: last-writer-wins on config, but bump the version DB-side
+        # (version = version + 1) so it stays strictly monotonic even when our ORM copy
+        # is stale. An ORM `+= 1` off a stale read could reuse a version a concurrent
+        # PATCH already consumed, letting a later guarded PATCH pass its version check.
+        replaced = await session.execute(
+            update(MCPServer)
+            .where(MCPServer.id == existing.id)
+            .values(
+                config=encrypt_mcp_config(server_config),
+                transport=_derive_transport(server_config),
+                version=MCPServer.version + 1,
+                updated_at=datetime.now(timezone.utc),
+            )
+        )
         await session.commit()
+        if replaced.rowcount == 1:
+            break
+        # Row deleted under us: re-read, and a still-missing row takes the create path
+        # next iteration — the replace converges to a re-create instead of silently
+        # returning None with the write lost.
+        session.expire(existing)
+        result = await session.exec(
+            select(MCPServer).where(MCPServer.user_id == user_id, MCPServer.name == server_name)
+        )
+        existing = result.first()
+    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

@@ -11,6 +11,7 @@
 """
 
 import asyncio
+import sqlite3
 import uuid
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
@@ -329,3 +330,162 @@ 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_patch_losing_to_concurrent_delete_returns_404(tmp_path):
+    """A merge PATCH whose row is deleted between its read and its guarded write returns 404, not a raw DB error."""
+    from fastapi import HTTPException
+
+    engine = await _file_engine(tmp_path / "mcp.db")
+    db_path = str(tmp_path / "mcp.db")
+    user = SimpleNamespace(id=uuid.uuid4())
+
+    import langflow.api.v2.mcp as mcp_mod
+
+    real_decrypt = mcp_mod.decrypt_mcp_config
+    fired = {"done": False}
+
+    def decrypt_then_delete(config):
+        # The PATCH calls decrypt between its read and its version-guarded UPDATE. Delete
+        # the row right there (raw sqlite = synchronous) so the guarded UPDATE matches 0
+        # rows and the re-read finds nothing -> a clean 404 rather than a raw DB error.
+        if not fired["done"]:
+            fired["done"] = True
+            con = sqlite3.connect(db_path, timeout=30)
+            con.execute("DELETE FROM mcp_server WHERE name = 'svc'")
+            con.commit()
+            con.close()
+        return real_decrypt(config)
+
+    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", "a": "1"}, user, session, None, None)
+
+        with patch.object(mcp_mod, "decrypt_mcp_config", decrypt_then_delete):
+            async with AsyncSession(engine, expire_on_commit=False) as session:
+                with pytest.raises(HTTPException) as exc:
+                    await update_server("svc", {"b": "2"}, user, session, None, None, merge_existing=True)
+
+    assert exc.value.status_code == 404, f"expected 404, got {exc.value.status_code}"
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        gone = (await session.exec(select(MCPServer).where(MCPServer.name == "svc"))).first()
+    await engine.dispose()
+    assert gone is None, "DELETE should have removed the row"
+
+
+@pytest.mark.asyncio
+async def test_full_replace_bumps_version_db_side_under_concurrent_write(tmp_path):
+    """Replace racing a concurrent write must bump the version DB-side (monotonic), not reuse a stale value.
+
+    The bug only fires when another writer advances the row between the replace's own read and its write
+    (a sequential PATCH-then-replace re-reads fresh and can't trigger it). So we inject the concurrent
+    version advance at exactly that point - inside `_derive_transport`, which the replace calls between its
+    read and its UPDATE - via a synchronous raw-sqlite write (a concurrent PATCH's effect on the version).
+    A naive ORM `+= 1` off the stale read would write 2 (reused); a DB-side `version + 1` writes 6.
+    """
+    engine = await _file_engine(tmp_path / "mcp.db")
+    db_path = str(tmp_path / "mcp.db")
+    user = SimpleNamespace(id=uuid.uuid4())
+
+    import langflow.api.v2.mcp as mcp_mod
+
+    real_derive = mcp_mod._derive_transport
+    raced = {"done": False}
+
+    def derive_then_advance(config):
+        if not raced["done"]:
+            raced["done"] = True
+            con = sqlite3.connect(db_path, timeout=30)
+            con.execute("UPDATE mcp_server SET version = 5 WHERE name = 'svc'")
+            con.commit()
+            con.close()
+        return real_derive(config)
+
+    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"}, user, session, None, None)  # version 1
+        with patch.object(mcp_mod, "_derive_transport", derive_then_advance):
+            async with AsyncSession(engine, expire_on_commit=False) as session:
+                await update_server("svc", {"url": "https://y", "replaced": "yes"}, user, session, None, None)
+        async with AsyncSession(engine, expire_on_commit=False) as session:
+            row = (await session.exec(select(MCPServer).where(MCPServer.name == "svc"))).first()
+
+    await engine.dispose()
+    assert row.version == 6, f"replace must bump DB-side (5 + 1 = 6), got {row.version} (stale/reused version)"
+    assert row.config.get("replaced") == "yes", "replace must overwrite the config"
+
+
+@pytest.mark.asyncio
+async def test_replace_losing_to_concurrent_delete_recreates(tmp_path):
+    """A full replace whose row is deleted between its read and its UPDATE re-creates the row.
+
+    Without the rowcount check the unguarded UPDATE matches nothing, the loop breaks, and the
+    request returns None with the write silently lost (200, no row). With it, the re-read finds
+    no row and the next iteration takes the create path, so the replace converges to a re-create.
+    """
+    engine = await _file_engine(tmp_path / "mcp.db")
+    db_path = str(tmp_path / "mcp.db")
+    user = SimpleNamespace(id=uuid.uuid4())
+
+    import langflow.api.v2.mcp as mcp_mod
+
+    real_derive = mcp_mod._derive_transport
+    raced = {"done": False}
+
+    def derive_then_delete(config):
+        # The replace calls _derive_transport between its read and its unguarded UPDATE;
+        # delete the row right there (raw sqlite = synchronous) so the UPDATE matches 0 rows.
+        if not raced["done"]:
+            raced["done"] = True
+            con = sqlite3.connect(db_path, timeout=30)
+            con.execute("DELETE FROM mcp_server WHERE name = 'svc'")
+            con.commit()
+            con.close()
+        return real_derive(config)
+
+    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", "old": "1"}, user, session, None, None)
+        with patch.object(mcp_mod, "_derive_transport", derive_then_delete):
+            async with AsyncSession(engine, expire_on_commit=False) as session:
+                result = await update_server("svc", {"url": "https://y", "replaced": "yes"}, user, session, None, None)
+        async with AsyncSession(engine, expire_on_commit=False) as session:
+            row = (await session.exec(select(MCPServer).where(MCPServer.name == "svc"))).first()
+
+    await engine.dispose()
+    assert row is not None, "replace raced by a delete must re-create the row, not silently no-op"
+    assert row.version == 1, f"re-created row starts fresh at version 1, got {row.version}"
+    assert row.config.get("replaced") == "yes", "re-created row must carry the replace's config"
+    assert row.config.get("old") is None, "re-created row must not resurrect the deleted config"
+    assert result == {"url": "https://y", "replaced": "yes"}, f"replace must return the written config, got {result!r}"