fix(mcp): re-create when a full replace races a concurrent delete

The merge-PATCH path already turns a row deleted between its read and its
guarded UPDATE into a clean 404. The full-replace path never checked
rowcount: the same race made its unguarded UPDATE match nothing, the loop
broke, and the request returned None with the write silently lost.

Check rowcount after the replace UPDATE; on a miss, re-read the row and
loop. A still-missing row takes the create path on the next iteration, so
the replace converges to a re-create (fresh row, version 1) instead of a
silent no-op. Covered by a race-injection test in the same style as
test_patch_losing_to_concurrent_delete_returns_404.

Author: erichare

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

@@ -506,7 +506,7 @@ async def update_server(
         # (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.
-        await session.execute(
+        replaced = await session.execute(
             update(MCPServer)
             .where(MCPServer.id == existing.id)
             .values(
@@ -517,7 +517,16 @@ async def update_server(
             )
         )
         await session.commit()
-        break
+        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.")
 

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

@@ -444,3 +444,48 @@ def derive_then_advance(config):
     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}"