From e277b4b1cbd01f0eb1085e12201d91a42cf8a8a8 Mon Sep 17 00:00:00 2001 From: Eric Hare Date: Mon, 27 Apr 2026 09:35:58 -0700 Subject: [PATCH] fix: migrate orphaned MCP servers config across DB resets (#12762) * 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. --- src/backend/base/langflow/services/utils.py | 149 +++++++++++ .../unit/test_mcp_servers_orphan_migration.py | 247 ++++++++++++++++++ 2 files changed, 396 insertions(+) create mode 100644 src/backend/tests/unit/test_mcp_servers_orphan_migration.py diff --git a/src/backend/base/langflow/services/utils.py b/src/backend/base/langflow/services/utils.py index f40bb89572..4a901496f2 100644 --- a/src/backend/base/langflow/services/utils.py +++ b/src/backend/base/langflow/services/utils.py @@ -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 diff --git a/src/backend/tests/unit/test_mcp_servers_orphan_migration.py b/src/backend/tests/unit/test_mcp_servers_orphan_migration.py new file mode 100644 index 0000000000..f0a1e7857a --- /dev/null +++ b/src/backend/tests/unit/test_mcp_servers_orphan_migration.py @@ -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