fix(review): address critical issues found in PR #12783 code review

preload.py — fork-safety bug: if _engine.dispose() raised, cache_service.
teardown() was skipped entirely, leaving the Redis/cache socket open and
inherited by all forked workers. Use separate try/finally so both teardowns
always run; re-raise the first error at the end.

server.py — _langflow_post_fork had two completely silent except blocks
(pass with no logging). Replace with server.log.warning(..., exc_info=True)
so fork-safety reset failures are visible in production logs.

alembic/env.py — added _db_models.import_all() before target_metadata is
read. The models package uses lazy __getattr__; out-of-band `alembic upgrade
head` (CLI, docker entrypoint) bypassed _run_migrations, saw an empty
SQLModel.metadata, and would generate spurious DROP TABLE operations.

models/__init__.py — fixed import_all() docstring: "avoid loading pandas"
was wrong (no model submodule imports pandas). Updated to describe the actual
reason (cold-start import cost) and document alembic/env.py as a call site.

starter_project_hash.py — docstring said main.py owns the hash gate; the
actual call site is preload.py.

components.py — three fixes:
  1. asyncio.Lock() comment claimed it raises RuntimeError on Python 3.13+.
     This is factually wrong (Lock() has not required a running loop since
     3.10). Replaced with the accurate reason: post_fork reset contract.
     Also removed an orphaned sentence that contradicted the preceding text.
  2. Cache-path resolution failure was logged at adebug/debug in two places
     (invisible at default log levels, silently defeats the cache on every
     cold start). Both raised to warning.

Author: jordanrfrazier

src/backend/base/langflow/alembic/env.py

@@ -1,20 +1,24 @@
-# noqa: INP001
 import asyncio
 import hashlib
 import os
 from logging.config import fileConfig
 from typing import Any
 
-
-from alembic import context
+from lfx.log.logger import logger
 from sqlalchemy import pool, text
 from sqlalchemy.event import listen
 from sqlalchemy.ext.asyncio import async_engine_from_config
 
-from lfx.log.logger import logger
-
+from alembic import context
+from langflow.services.database import models as _db_models
 from langflow.services.database.service import SQLModel
 
+# Eagerly import all SQLModel subclasses so they register in SQLModel.metadata
+# before Alembic reads target_metadata.  The models package uses a lazy
+# __getattr__ to avoid cold-start import cost, so without this call any
+# out-of-band `alembic upgrade head` (CLI, docker entrypoint, etc.) would
+# see an empty/partial metadata and generate spurious DROP TABLE operations.
+_db_models.import_all()
 
 # this is the Alembic Config object, which provides
 # access to the values within the .ini file in use.

src/backend/base/langflow/initial_setup/starter_project_hash.py

@@ -2,8 +2,8 @@
 
 Computes a content hash over the starter-project JSON files plus the installed
 ``lfx`` package version, and persists it as plaintext under
-``${LANGFLOW_CONFIG_DIR}/starter_projects.hash``. ``main.py`` uses the hash to
-short-circuit the full starter-project re-sync on restarts where nothing
+``${LANGFLOW_CONFIG_DIR}/starter_projects.hash``. ``preload.py`` uses the hash
+to short-circuit the full starter-project re-sync on restarts where nothing
 changed.
 
 Failure modes: missing, unreadable, or corrupt hash files all fall

src/backend/base/langflow/preload.py

@@ -327,21 +327,29 @@ async def _run_agentic_mcp() -> None:
         # usable: on first access in a worker it opens a fresh pool for that
         # process.
         #
-        # Guard: if initialize_services raised before registering the DB service
-        # we skip disposal (nothing to close). We check the service registry
-        # directly so that a real dispose() failure propagates instead of being
-        # swallowed — a failed dispose() is a fork-safety hazard.
+        # CRITICAL: both DB engine disposal and cache service teardown are
+        # fork-safety-critical — both must run even if one raises.  Use
+        # separate try/finally so a dispose() failure does not skip teardown()
+        # (and vice versa), then re-raise the first error.
         await logger.adebug("[preload] disposing master DB engine before fork")
         from langflow.services.manager import get_service_manager
         from langflow.services.schema import ServiceType as _ServiceType
 
-        if _ServiceType.DATABASE_SERVICE in get_service_manager().services:
-            _db_svc = get_db_service()
-            _engine = getattr(_db_svc, "engine", None)
-            if _engine is not None:
-                await _engine.dispose()
-        else:
-            await logger.adebug("[preload] DB engine dispose skipped (service not yet initialized)")
+        _db_dispose_err: BaseException | None = None
+        try:
+            # Guard: if initialize_services raised before registering the DB
+            # service we skip disposal (nothing to close).  We check the
+            # service registry directly so that a real dispose() failure
+            # propagates — a failed dispose() is a fork-safety hazard.
+            if _ServiceType.DATABASE_SERVICE in get_service_manager().services:
+                _db_svc = get_db_service()
+                _engine = getattr(_db_svc, "engine", None)
+                if _engine is not None:
+                    await _engine.dispose()
+            else:
+                await logger.adebug("[preload] DB engine dispose skipped (service not yet initialized)")
+        except BaseException as _exc:  # noqa: BLE001
+            _db_dispose_err = _exc
 
         # Close cache service socket (e.g. Redis) to prevent sharing across fork.
         # ExternalAsyncBaseCacheService declares teardown() abstract, so any
@@ -355,6 +363,9 @@ async def _run_agentic_mcp() -> None:
         if isinstance(cache_service, ExternalAsyncBaseCacheService):
             await cache_service.teardown()
 
+        if _db_dispose_err is not None:
+            raise _db_dispose_err
+
 
 def preload_master() -> None:
     """Run one-time Langflow initialization in the gunicorn master before workers are forked.

src/backend/base/langflow/server.py

@@ -192,15 +192,24 @@ def _langflow_post_fork(server, worker) -> None:  # noqa: ARG001
         from langflow.services.deps import get_telemetry_service
 
         get_telemetry_service().client = None
-    except Exception:  # noqa: BLE001, S110
+    except Exception:  # noqa: BLE001
         # Service not yet initialized (e.g. preload_app=False path). The
-        # hook must not crash gunicorn.
-        pass
+        # hook must not crash gunicorn, but the failure is worth knowing about.
+        server.log.warning(
+            "[post_fork] Failed to reset TelemetryService.client; "
+            "worker telemetry may fail on first use",
+            exc_info=True,
+        )
 
     try:
         from lfx.interface.components import component_cache
 
         component_cache._lock = None  # noqa: SLF001
-    except Exception:  # noqa: BLE001, S110
-        # Module not importable in this worker for any reason — non-fatal.
-        pass
+    except Exception:  # noqa: BLE001
+        # Module not importable in this worker for any reason — non-fatal,
+        # but log so operators can diagnose deployment issues.
+        server.log.warning(
+            "[post_fork] Failed to reset component_cache._lock; "
+            "worker may inherit stale asyncio.Lock from master",
+            exc_info=True,
+        )

src/backend/base/langflow/services/database/models/__init__.py

@@ -66,11 +66,13 @@ def import_all() -> None:
 
     Call this before running Alembic migrations or any code that iterates
     __dict__ to discover SQLModel subclasses (e.g. schema health checks).
-    Models are lazy by default to avoid loading pandas at package import time.
+    Models are lazy by default to reduce cold-start import cost (each submodule
+    pulls in SQLAlchemy validators, pydantic field definitions, etc.).
 
-    All call sites have been audited: only database/service.py iterates __dict__
-    for model discovery, and both call sites (migration and health-check paths)
-    call import_all() immediately before the iteration.
+    Call sites: database/service.py (migration + health-check paths) and
+    alembic/env.py (so out-of-band `alembic upgrade` sees complete metadata).
+    All call sites must call import_all() before iterating SQLModel.metadata
+    or __dict__ for model discovery.
     """
     for name in __all__:
         getattr(__import__(__name__, fromlist=[name]), name)

src/lfx/src/lfx/interface/components.py

@@ -48,18 +48,22 @@ def __init__(self):
         # None means "not yet loaded" (fail-closed); {} means "loaded, no components found".
         self.type_to_current_hash: dict[str, set[str]] | None = None
         self.all_known_hashes: set[str] | None = None
-        # Lazily created on first access from inside a running event loop.
-        # Constructing asyncio.Lock() at import time raises RuntimeError on
-        # Python 3.13+ because the singleton below runs at module import.
+        # Lazily created to avoid binding event-loop state at import time and
+        # to allow the post_fork hook to reset it to None so each worker
+        # creates a fresh Lock against its own event loop (see server.py
+        # _langflow_post_fork).
         self._lock: asyncio.Lock | None = None
 
     @property
     def lock(self) -> asyncio.Lock:
         """Return the asyncio.Lock, creating it lazily on first access.
 
-        Must be called from inside a running event loop: asyncio.Lock() requires
-        a running loop on Python 3.13+, and the singleton is instantiated at
-        module import time.
+        The lock is created on first use (not at import time) so that the
+        post_fork hook in server.py can reset it to None, causing each worker
+        to construct a fresh Lock bound to its own event loop on the next
+        access.  Creating it eagerly in __init__ would bind it to the master's
+        event loop, causing "Future attached to a different loop" errors in
+        workers.
         """
         if self._lock is None:
             self._lock = asyncio.Lock()
@@ -380,7 +384,10 @@ async def _load_from_index_or_cache(
     try:
         cache_path = _get_cache_path()
     except Exception as e:  # noqa: BLE001
-        await logger.adebug(f"Cache load failed: {e}")
+        await logger.awarning(
+            f"Could not determine component cache path ({type(e).__name__}: {e}); "
+            "falling through to dynamic rebuild"
+        )
     else:
         if cache_path.exists():
             await logger.adebug(f"Attempting to load from cache: {cache_path}")
@@ -763,7 +770,10 @@ async def get_and_cache_all_types_dict(
             try:
                 cache_path = _get_cache_path()
             except Exception as exc:  # noqa: BLE001
-                logger.debug(f"Could not resolve component cache path: {exc}")
+                logger.warning(
+                    f"Could not resolve component cache path ({type(exc).__name__}: {exc}); "
+                    "falling through to dynamic rebuild"
+                )
                 cache_path = None
             if cache_path is not None and cache_path.exists():
                 cached_blob: Any = None