From 075a3bce72fcc9d9f787493ee33aaea6ac3377ff Mon Sep 17 00:00:00 2001 From: Arek Mateusiak Date: Tue, 21 Apr 2026 19:09:52 +0200 Subject: [PATCH] feat(preload): Introduce per-step completion flags for improved state management - Added completion flags in the preload state to track the status of various initialization steps, including profile picture copying, starter project creation, agentic global variable initialization, MCP server configuration, and flow loading. - Updated the lifespan management in `main.py` to utilize these flags, allowing the system to skip redundant setup tasks if they have already been completed during the preload phase. - This enhancement improves resource management and ensures that the application behaves correctly in a multi-worker environment. --- src/backend/base/langflow/main.py | 22 +++++++++++----------- src/backend/base/langflow/preload.py | 12 ++++++++++++ 2 files changed, 23 insertions(+), 11 deletions(-) diff --git a/src/backend/base/langflow/main.py b/src/backend/base/langflow/main.py index 622803adac..e80e558922 100644 --- a/src/backend/base/langflow/main.py +++ b/src/backend/base/langflow/main.py @@ -153,7 +153,7 @@ def get_lifespan(*, fix_migration=False, version=None): async def lifespan(_app: FastAPI): from lfx.interface.components import get_and_cache_all_types_dict - from langflow.preload import get_preloaded_temp_dirs, is_master, is_preloaded + from langflow.preload import _STATE, get_preloaded_temp_dirs, is_master, is_preloaded preloaded = is_preloaded() running_in_master = is_master() @@ -208,7 +208,7 @@ def get_lifespan(*, fix_migration=False, version=None): setup_llm_caching() await logger.adebug(f"LLM caching setup in {asyncio.get_event_loop().time() - current_time:.2f}s") - if not preloaded: + if not preloaded or not _STATE.profile_pictures_copied: current_time = asyncio.get_event_loop().time() await logger.adebug("Copying profile pictures") await copy_profile_pictures() @@ -225,10 +225,8 @@ def get_lifespan(*, fix_migration=False, version=None): await logger.adebug("Initializing super user") await initialize_auto_login_default_superuser() await logger.adebug(f"Super user initialized in {asyncio.get_event_loop().time() - current_time:.2f}s") - else: - await logger.adebug( - "Skipping profile-picture copy and super user setup: master already ran them during preload" - ) + elif _STATE.profile_pictures_copied: + await logger.adebug("Skipping profile-picture copy: master already completed it during preload") if get_settings_service().settings.prometheus_enabled: try: @@ -255,7 +253,7 @@ def get_lifespan(*, fix_migration=False, version=None): telemetry_service = get_telemetry_service() - if preloaded: + if preloaded and _STATE.starter_projects_created: # Inherit bundle paths and types dict from master via COW. # Only the master owns the bundle temp dirs; workers must NOT # add them to their own cleanup list, or shutdown would delete @@ -299,7 +297,7 @@ def get_lifespan(*, fix_migration=False, version=None): ) # Initialize agentic global variables early (before MCP server and flows) - if get_settings_service().settings.agentic_experience: + if get_settings_service().settings.agentic_experience and not _STATE.agentic_globals_initialized: from langflow.api.utils.mcp.agentic_mcp import initialize_agentic_global_variables current_time = asyncio.get_event_loop().time() @@ -328,8 +326,8 @@ def get_lifespan(*, fix_migration=False, version=None): ) # Auto-configure Agentic MCP server if enabled (after variables are initialized). - # Skipped when preloaded: the master already seeded this into the DB. - if get_settings_service().settings.agentic_experience and not preloaded: + # Skipped when the master already completed this during preload. + if get_settings_service().settings.agentic_experience and not _STATE.agentic_mcp_configured: from langflow.api.utils.mcp.agentic_mcp import auto_configure_agentic_mcp_server current_time = asyncio.get_event_loop().time() @@ -344,9 +342,11 @@ def get_lifespan(*, fix_migration=False, version=None): await logger.awarning(f"Failed to configure agentic MCP server: {e}") current_time = asyncio.get_event_loop().time() - if not preloaded: + if not _STATE.flows_loaded: await logger.adebug("Loading flows") await load_flows_from_directory() + else: + await logger.adebug("Skipping flows load: master already completed it during preload") # ``sync_flows_from_fs`` and the queue service MUST be started # per-worker: they create asyncio tasks bound to this event loop. sync_flows_from_fs_task = asyncio.create_task(sync_flows_from_fs()) diff --git a/src/backend/base/langflow/preload.py b/src/backend/base/langflow/preload.py index 8027733364..ff9400b7d5 100644 --- a/src/backend/base/langflow/preload.py +++ b/src/backend/base/langflow/preload.py @@ -46,6 +46,12 @@ class _PreloadState: master_pid: int | None = None temp_dirs: list[TemporaryDirectory] = field(default_factory=list) bundles_components_paths: list[str] = field(default_factory=list) + # Per-step completion flags to prevent silent data loss + profile_pictures_copied: bool = False + starter_projects_created: bool = False + agentic_globals_initialized: bool = False + agentic_mcp_configured: bool = False + flows_loaded: bool = False _STATE = _PreloadState() @@ -106,6 +112,7 @@ async def _run_master_preload() -> None: await logger.adebug("[preload] copying profile pictures") try: await copy_profile_pictures() + _STATE.profile_pictures_copied = True except Exception as e: # noqa: BLE001 await logger.awarning(f"[preload] copy_profile_pictures failed: {e}") @@ -123,6 +130,7 @@ async def _run_master_preload() -> None: await logger.adebug("[preload] creating/updating starter projects") try: await create_or_update_starter_projects(all_types_dict) + _STATE.starter_projects_created = True except Exception as e: # noqa: BLE001 await logger.awarning(f"[preload] starter projects init failed: {e}") @@ -136,15 +144,19 @@ async def _run_master_preload() -> None: await logger.adebug("[preload] initializing agentic global variables") async with session_scope() as session: await initialize_agentic_global_variables(session) + _STATE.agentic_globals_initialized = True + await logger.adebug("[preload] auto-configuring agentic MCP server") async with session_scope() as session: await auto_configure_agentic_mcp_server(session) + _STATE.agentic_mcp_configured = True except Exception as e: # noqa: BLE001 await logger.awarning(f"[preload] agentic MCP init failed: {e}") await logger.adebug("[preload] loading flows from directory") try: await load_flows_from_directory() + _STATE.flows_loaded = True except Exception as e: # noqa: BLE001 await logger.awarning(f"[preload] load_flows_from_directory failed: {e}")