pengjunjie/langflow
1
0
Fork 1
mirror of https://github.com/langflow-ai/langflow.git synced 2026-07-24 06:10:49 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: migrate orphaned MCP servers config across DB resets (#12762)

Browse Source 
* fix: migrate orphaned MCP servers config across DB resets

When Langflow restarts with a fresh database but the same LANGFLOW_CONFIG_DIR
(common in containerized deployments without a persisted DB volume), the
default superuser is recreated with a new UUID and the previously saved
_mcp_servers_{old_uuid}.json files become unreachable.

On default-superuser creation in AUTO_LOGIN mode, scan the config directory
for orphaned _mcp_servers_*.json files under prior UUID folders and migrate
the most recently modified one to the new user (copy contents + register a
UserFile row). Never overwrites an existing file and is a no-op when no
orphan is found; failures are swallowed so migration cannot block startup.

Fixes langflow-ai/langflow#9524

* fix: address review feedback on MCP orphan migration

- Refuse to migrate when multiple orphan candidates are present. MCP server
  entries can contain env/headers auth material, so silently picking the
  newest orphan could import an unrelated user's config. Operators get a
  warning with candidate paths for manual recovery.
- Self-heal missing DB rows when the current user's MCP config file already
  exists on disk (e.g. from a previous migration that crashed before commit)
  instead of returning early and leaving the config invisible.
- Add regression tests for the multi-orphan skip path, self-heal path, and
  "DB row already present" short-circuit.
This commit is contained in:
Eric Hare
2026-04-27 09:35:58 -07:00
committed by GitHub
parent d8bad2c1db
commit e277b4b1cb
2 changed files with 396 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

149
src/backend/base/langflow/services/utils.py
View File

@ -94,6 +94,12 @@ async def setup_superuser(settings_service: SettingsService, session: AsyncSessi
)
if user is not None:
await logger.adebug("Superuser created successfully.")
# When the default superuser is recreated (e.g. after a DB reset in
# AUTO_LOGIN mode) the per-user MCP servers config file saved under the
# previous UUID becomes orphaned on disk. Best-effort recover it so
# users don't lose their MCP server configuration across restarts.
if is_default and settings_service.auth_settings.AUTO_LOGIN:
await migrate_orphaned_mcp_servers_config(session, settings_service, user)
except Exception as exc:
logger.exception(exc)
msg = "Could not create superuser. Please create a superuser manually."
@ -103,6 +109,149 @@ async def setup_superuser(settings_service: SettingsService, session: AsyncSessi
settings_service.auth_settings.reset_credentials()
async def migrate_orphaned_mcp_servers_config(
session: AsyncSession,
settings_service: SettingsService,
current_user,
) -> bool:
"""Best-effort recovery of MCP servers config files orphaned by a DB reset.
The MCP servers config is persisted on disk at
``{config_dir}/{user_id}/_mcp_servers_{user_id}.json`` and tracked in the DB
via a ``File`` row. When Langflow starts with a fresh database but the same
config directory (common in containerized deployments without a persisted
DB volume), the default superuser is recreated with a new UUID and the
previously saved MCP config files become unreachable.
Recovery rules (intentionally conservative to avoid importing another user's
config — MCP server entries can contain ``env`` and ``headers`` auth material):
* If the new user already has an MCP config file on disk without a matching
``File`` row, re-register the row (self-heal a partial previous migration).
* If exactly one orphaned ``_mcp_servers_{uuid}.json`` is found in the config
directory, migrate it.
* If multiple orphans are found, skip and log — we can't safely identify the
previous default superuser's file without extra metadata, so leave manual
recovery to the operator.
Returns True when a file was migrated or a missing DB row was restored.
"""
from pathlib import Path
from uuid import UUID
import aiofiles
import anyio
from langflow.services.database.models.file.model import File as UserFile
try:
config_dir_value = settings_service.settings.config_dir
if not config_dir_value:
return False
config_dir = Path(config_dir_value)
if not config_dir.exists() or not config_dir.is_dir():
return False
# The current user's DB record is fresh; nothing to migrate if they
# somehow already have an MCP config row (defensive guard).
name_without_ext = f"_mcp_servers_{current_user.id}"
existing_stmt = (
select(UserFile).where(UserFile.user_id == current_user.id).where(UserFile.name == name_without_ext)
)
if (await session.exec(existing_stmt)).first() is not None:
return False
current_user_dir = str(current_user.id)
new_dir = config_dir / current_user_dir
new_filename = f"_mcp_servers_{current_user.id}.json"
new_file_path = new_dir / new_filename
db_path = f"{current_user.id}/{new_filename}"
# Case 1: a previous migration attempt copied the file but failed before
# committing the DB row. Re-register the existing file instead of
# returning early and leaving the user with an invisible config.
if new_file_path.exists():
try:
size = new_file_path.stat().st_size
except OSError as exc:
await logger.awarning(
"Cannot stat existing MCP config %s while self-healing DB row: %s",
new_file_path,
exc,
)
return False
session.add(UserFile(user_id=current_user.id, name=name_without_ext, path=db_path, size=size))
await session.commit()
await logger.ainfo(
"Restored missing MCP servers config DB row for user %s from existing file %s",
current_user.id,
new_file_path,
)
return True
def _find_orphans() -> list[tuple[float, Path]]:
orphans: list[tuple[float, Path]] = []
for entry in config_dir.iterdir():
if not entry.is_dir() or entry.name == current_user_dir:
continue
try:
UUID(entry.name)
except ValueError:
continue
mcp_path = entry / f"_mcp_servers_{entry.name}.json"
if mcp_path.is_file():
try:
mtime = mcp_path.stat().st_mtime
except OSError:
continue
orphans.append((mtime, mcp_path))
return orphans
orphans = await anyio.to_thread.run_sync(_find_orphans)
if not orphans:
return False
# Ambiguous: more than one candidate could belong to different users.
# Refuse to migrate rather than risk importing unrelated auth material.
if len(orphans) > 1:
orphan_paths = ", ".join(str(p) for _, p in sorted(orphans, key=lambda i: i[0], reverse=True))
await logger.awarning(
"Found %d orphaned MCP servers config files in %s; skipping automatic "
"migration to avoid restoring the wrong one. Move the intended file to "
"%s to recover. Candidates: %s",
len(orphans),
config_dir,
new_file_path,
orphan_paths,
)
return False
_, orphan_path = orphans[0]
async with aiofiles.open(str(orphan_path), "rb") as src:
data = await src.read()
await anyio.to_thread.run_sync(lambda: new_dir.mkdir(parents=True, exist_ok=True))
async with aiofiles.open(str(new_file_path), "wb") as dst:
await dst.write(data)
session.add(UserFile(user_id=current_user.id, name=name_without_ext, path=db_path, size=len(data)))
await session.commit()
await logger.ainfo(
"Migrated orphaned MCP servers config from %s to user %s",
orphan_path,
current_user.id,
)
except Exception as exc: # noqa: BLE001
# Never let migration failure block startup.
await logger.awarning("Failed to migrate orphaned MCP servers config: %s", exc)
return False
else:
return True
async def teardown_superuser(settings_service, session: AsyncSession) -> None:
"""Teardown the superuser."""
# If AUTO_LOGIN is True, we will remove the default superuser

247
src/backend/tests/unit/test_mcp_servers_orphan_migration.py Normal file
View File

@ -0,0 +1,247 @@
"""Tests for migrate_orphaned_mcp_servers_config in langflow.services.utils.
Verifies that MCP server config files written under a previous default
superuser's UUID are picked up and migrated to the new default superuser
when the database is reset but the config directory is preserved
(typical of containerized deployments without a persisted DB volume).
"""
from __future__ import annotations
import json
import os
from typing import TYPE_CHECKING
from uuid import uuid4
import pytest
if TYPE_CHECKING:
from pathlib import Path
from langflow.services.database.models.file.model import File as UserFile
from langflow.services.deps import get_settings_service, session_scope
from langflow.services.utils import migrate_orphaned_mcp_servers_config
from sqlmodel import select
@pytest.fixture
async def initialized_services(monkeypatch, tmp_path):
"""Initialize DB + services with an isolated config dir."""
from langflow.services.utils import initialize_services, teardown_services
from lfx.services.manager import get_service_manager
db_path = tmp_path / "test.db"
config_dir = tmp_path / "config"
config_dir.mkdir(parents=True, exist_ok=True)
monkeypatch.setenv("LANGFLOW_DATABASE_URL", f"sqlite:///{db_path}")
monkeypatch.setenv("LANGFLOW_CONFIG_DIR", str(config_dir))
monkeypatch.setenv("LANGFLOW_AUTO_LOGIN", "true")
get_service_manager().factories.clear()
get_service_manager().services.clear()
await initialize_services()
yield config_dir
await teardown_services()
def _write_orphan(config_dir: Path, payload: dict, *, mtime_offset: float = 0.0) -> Path:
"""Create an orphaned _mcp_servers_{uuid}.json file in a UUID-named folder."""
orphan_id = uuid4()
orphan_dir = config_dir / str(orphan_id)
orphan_dir.mkdir(parents=True, exist_ok=True)
orphan_path = orphan_dir / f"_mcp_servers_{orphan_id}.json"
orphan_path.write_text(json.dumps(payload))
if mtime_offset:
stat = orphan_path.stat()
os.utime(orphan_path, (stat.st_atime + mtime_offset, stat.st_mtime + mtime_offset))
return orphan_path
@pytest.mark.asyncio
@pytest.mark.timeout(30)
async def test_migrate_orphaned_mcp_servers_config_recovers_previous_user_config(
initialized_services,
):
"""A single orphaned file is migrated to the current default superuser."""
config_dir: Path = initialized_services
expected_payload = {"mcpServers": {"my-server": {"command": "uvx", "args": ["mcp-proxy"]}}}
orphan_path = _write_orphan(config_dir, expected_payload)
settings = get_settings_service()
async with session_scope() as session:
from langflow.services.database.models.user.model import User
from lfx.services.settings.constants import DEFAULT_SUPERUSER
user = (await session.exec(select(User).where(User.username == DEFAULT_SUPERUSER))).first()
assert user is not None, "default superuser should exist after initialize_services"
# Simulate the fresh-DB scenario: the user has no MCP config row yet.
stmt = select(UserFile).where(UserFile.user_id == user.id).where(UserFile.name == f"_mcp_servers_{user.id}")
assert (await session.exec(stmt)).first() is None
migrated = await migrate_orphaned_mcp_servers_config(session, settings, user)
assert migrated is True
# DB record should exist and point at the new user-specific path.
stmt = select(UserFile).where(UserFile.user_id == user.id).where(UserFile.name == f"_mcp_servers_{user.id}")
new_record = (await session.exec(stmt)).first()
assert new_record is not None
assert new_record.path == f"{user.id}/_mcp_servers_{user.id}.json"
# File should live under the new user's folder with the same contents.
target = config_dir / str(user.id) / f"_mcp_servers_{user.id}.json"
assert target.exists()
assert json.loads(target.read_text()) == expected_payload
# Orphan source is left intact (best-effort copy, not destructive move).
assert orphan_path.exists()
@pytest.mark.asyncio
@pytest.mark.timeout(30)
async def test_migrate_orphaned_mcp_servers_config_skips_when_multiple_orphans(
initialized_services,
):
"""With multiple orphan candidates we refuse to guess and leave state untouched.
MCP server entries can contain env/headers auth material, so importing an
unrelated user's config would be a security hazard. Operators must resolve
the ambiguity manually.
"""
config_dir: Path = initialized_services
old_payload = {"mcpServers": {"old": {}}}
new_payload = {"mcpServers": {"new": {}}}
old_path = _write_orphan(config_dir, old_payload, mtime_offset=-3600)
new_path = _write_orphan(config_dir, new_payload)
settings = get_settings_service()
async with session_scope() as session:
from langflow.services.database.models.user.model import User
from lfx.services.settings.constants import DEFAULT_SUPERUSER
user = (await session.exec(select(User).where(User.username == DEFAULT_SUPERUSER))).first()
migrated = await migrate_orphaned_mcp_servers_config(session, settings, user)
assert migrated is False
stmt = select(UserFile).where(UserFile.user_id == user.id).where(UserFile.name == f"_mcp_servers_{user.id}")
assert (await session.exec(stmt)).first() is None
target = config_dir / str(user.id) / f"_mcp_servers_{user.id}.json"
assert not target.exists()
# Both orphans are preserved on disk for manual recovery.
assert json.loads(old_path.read_text()) == old_payload
assert json.loads(new_path.read_text()) == new_payload
@pytest.mark.asyncio
@pytest.mark.timeout(30)
async def test_migrate_orphaned_mcp_servers_config_no_orphans_is_noop(
initialized_services,
):
"""With no orphaned files the function returns False and does not touch the DB."""
config_dir: Path = initialized_services
# No UUID-named subdirectories should exist.
from uuid import UUID
def _is_uuid_dir(p: Path) -> bool:
if not p.is_dir():
return False
try:
UUID(p.name)
except ValueError:
return False
return True
assert not [p for p in config_dir.iterdir() if _is_uuid_dir(p)]
settings = get_settings_service()
async with session_scope() as session:
from langflow.services.database.models.user.model import User
from lfx.services.settings.constants import DEFAULT_SUPERUSER
user = (await session.exec(select(User).where(User.username == DEFAULT_SUPERUSER))).first()
migrated = await migrate_orphaned_mcp_servers_config(session, settings, user)
assert migrated is False
stmt = select(UserFile).where(UserFile.user_id == user.id).where(UserFile.name == f"_mcp_servers_{user.id}")
assert (await session.exec(stmt)).first() is None
@pytest.mark.asyncio
@pytest.mark.timeout(30)
async def test_migrate_orphaned_mcp_servers_config_self_heals_missing_db_row(
initialized_services,
):
"""Re-register missing DB rows when the on-disk file already exists.
If the user's file is on disk but the DB row is missing, the migration
should recreate the row instead of leaving the config invisible. This
covers recovery from a previous migration attempt that wrote the file
but crashed before committing the DB row.
"""
config_dir: Path = initialized_services
settings = get_settings_service()
async with session_scope() as session:
from langflow.services.database.models.user.model import User
from lfx.services.settings.constants import DEFAULT_SUPERUSER
user = (await session.exec(select(User).where(User.username == DEFAULT_SUPERUSER))).first()
# Simulate a file-exists / DB-row-missing state.
existing_dir = config_dir / str(user.id)
existing_dir.mkdir(parents=True, exist_ok=True)
existing_payload = {"mcpServers": {"keep-me": {}}}
existing_path = existing_dir / f"_mcp_servers_{user.id}.json"
existing_path.write_text(json.dumps(existing_payload))
migrated = await migrate_orphaned_mcp_servers_config(session, settings, user)
assert migrated is True
stmt = select(UserFile).where(UserFile.user_id == user.id).where(UserFile.name == f"_mcp_servers_{user.id}")
row = (await session.exec(stmt)).first()
assert row is not None
assert row.path == f"{user.id}/_mcp_servers_{user.id}.json"
assert row.size == existing_path.stat().st_size
# File contents are untouched.
assert json.loads(existing_path.read_text()) == existing_payload
@pytest.mark.asyncio
@pytest.mark.timeout(30)
async def test_migrate_orphaned_mcp_servers_config_skips_when_row_already_present(
initialized_services,
):
"""When the DB row already exists, do nothing — avoids double-registration."""
config_dir: Path = initialized_services
_write_orphan(config_dir, {"mcpServers": {"orphan": {}}})
settings = get_settings_service()
async with session_scope() as session:
from langflow.services.database.models.user.model import User
from lfx.services.settings.constants import DEFAULT_SUPERUSER
user = (await session.exec(select(User).where(User.username == DEFAULT_SUPERUSER))).first()
# Pre-register an MCP file row to simulate an already-migrated user.
session.add(
UserFile(
user_id=user.id,
name=f"_mcp_servers_{user.id}",
path=f"{user.id}/_mcp_servers_{user.id}.json",
size=0,
)
)
await session.commit()
migrated = await migrate_orphaned_mcp_servers_config(session, settings, user)
assert migrated is False