UN-4136 [FIX] Stop the file_processing worker booting the executor stack on each child's first task

workers/executor/__init__.py

@@ -3,10 +3,33 @@
 Celery worker for running extraction executors.
 Dispatches ExecutionContext to registered executors and returns
 ExecutionResult via the Celery result backend.
+
+``celery_app`` resolves lazily (PEP 562). Importing it eagerly made *every*
+consumer of this package pay the executor worker's full bootstrap: ``.worker``
+built the Celery app, and registration then imported ``LegacyExecutor`` and the
+adapter stack behind it. That was ~9s of work the file_processing worker never
+needed — it imports ``ExecutorToolShim`` and some string constants from this
+package and dispatches everything else over the PG queue (UN-4136). Attribute
+access is unchanged, so ``from executor import celery_app`` still works for
+callers that genuinely want the app.
 """
 
-from .worker import app as celery_app
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from celery import Celery
+
+    celery_app: Celery
 
 __all__ = [
     "celery_app",
 ]
+
+
+def __getattr__(name: str) -> object:
+    """Resolve ``celery_app`` on first access instead of at import time."""
+    if name == "celery_app":
+        from .worker import app
+
+        return app
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

workers/executor/executors/__init__.py

@@ -1,16 +1,133 @@
 """Executor implementations package.
 
-Importing this module triggers ``@ExecutorRegistry.register`` for all
-bundled executors and discovers cloud executors via entry points.
+Registration is **explicit**: call :func:`register_all`. It used to run as a
+side effect of importing this package, which meant that importing *any*
+submodule — including ``executor.executors.constants``, which itself imports
+nothing but ``enum`` — dragged in ``LegacyExecutor`` and the entire adapter
+stack behind it. That cost ~9s, and the file_processing worker paid it once per
+forked child just to read a few string constants (UN-4136).
+
+``executor/tasks.py`` is the only *production* caller: ``workers/worker.py``
+exec-loads that file by path for both the Celery and PG executor roles.
+``executor/worker.py`` reaches it transitively, by importing ``executor.tasks``
+— that import binds the task definitions to its app and populates the registry,
+so it is load-bearing despite its ``noqa: F401``. The app it builds is not on
+any deployed path; today only the tests use it. The test suite calls
+:func:`register_all` directly.
 """
 
-from executor.executors.legacy_executor import LegacyExecutor
-from executor.executors.plugins.loader import ExecutorPluginLoader
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from executor.executors.legacy_executor import LegacyExecutor  # noqa: TCH004
+
+#: Cloud entry point names; None when discovery has not run, or ran and
+#: raised. Doubles as the re-entrancy latch — see :func:`register_all`.
+_cloud_executors: list[str] | None = None
+
+
+def register_all() -> list[str]:
+    """Import the executor modules once per process, registering each of them.
+
+    ``LegacyExecutor`` and every cloud executor carry
+    ``@ExecutorRegistry.register``, which fires when their module is first
+    imported. That is the whole mechanism, and it bounds the guarantee: this
+    populates ``ExecutorRegistry`` **in a fresh process**, and cannot repopulate
+    it if something empties it afterwards, because the second import is a
+    ``sys.modules`` hit and the decorator does not run again. Production never
+    clears the registry. Several test modules do, and most of them do not put it
+    back; ``tests/test_legacy_executor_scaffold.py`` restores what it cleared.
+    A module that re-registers must not call ``ExecutorRegistry.register`` while
+    the name is already present — either clear immediately before, or guard on
+    ``"legacy" not in ExecutorRegistry.list_executors()``. That rule is hand-copied
+    across the test modules in several spellings; folding it into a shared test
+    helper is deliberately left out of this change.
+
+    Idempotent, and safe to re-enter: a cloud plugin whose own import graph
+    reaches this function during ``ep.load()`` will not restart discovery.
+
+    Returns:
+        The cloud executor entry point names, as a fresh copy each time so a
+        caller cannot mutate the latched state.
+
+        The list can under-report the registry at any length, including zero.
+        An empty one does not distinguish its causes: no cloud plugins are
+        installed (the OSS case); every one of them failed to import, because
+        ``ExecutorPluginLoader.discover_executors`` catches per-entry-point
+        failures and only logs a warning, so a broken plugin wheel boots clean
+        (a follow-up to make that aggregate loud is recorded on UN-4136); the
+        caller is *inside* discovery, having re-entered while the latch is still
+        the empty placeholder; or a failed discovery was re-run and the plugin
+        that raised is live in ``ExecutorRegistry`` but absent here — see the
+        handler below.
+    """
+    global _cloud_executors
+
+    from executor.executors.legacy_executor import LegacyExecutor  # noqa: F401
+
+    if _cloud_executors is None:
+        from executor.executors.plugins.loader import ExecutorPluginLoader
+
+        # Latch BEFORE discovering. ``ep.load()`` executes third-party code, and
+        # a plugin that reaches back into this function would otherwise find the
+        # latch still unset and restart the entry point loop, nesting once per
+        # level. The import-side-effect version this replaced got that safety
+        # free from ``sys.modules``.
+        #
+        try:
+            _cloud_executors = []
+            _cloud_executors = ExecutorPluginLoader.discover_executors()
+        except BaseException:
+            # Re-raise unchanged, and put the latch back to its un-armed value,
+            # so a later call re-runs discovery instead of reporting "no cloud
+            # executors" as though this one had succeeded. That does re-arm a
+            # retry for a later caller — no production caller survives the
+            # re-raise to make one (see the module docstring), so today only
+            # tests reach it.
+            #
+            # ``BaseException`` rather than ``Exception``: see
+            # ``tests/test_plugin_loader.py::test_failed_discovery_un_arms_the_latch``,
+            # parametrised over every class that reaches here and pinning them
+            # executably.
+            #
+            # A re-run is not free. The plugin whose module was mid-import when
+            # the exception escaped is evicted from ``sys.modules`` but stays in
+            # the registry if its decorator had already fired, so re-importing
+            # it raises a duplicate-name ``ValueError`` that
+            # ``discover_executors`` swallows into a warning — live in the
+            # registry, absent from the returned list. Re-running discovery has
+            # always had that property, including in the import-side-effect
+            # version this replaced.
+            _cloud_executors = None
+            raise
+
+    return list(_cloud_executors)
+
+
+def __getattr__(name: str) -> object:
+    """Keep ``from executor.executors import LegacyExecutor`` working.
+
+    The class used to sit in this namespace as a side effect of the eager
+    import. Resolving it on access preserves that spelling without reviving the
+    cost for everyone who imports a sibling submodule.
+    """
+    if name == "LegacyExecutor":
+        from executor.executors.legacy_executor import LegacyExecutor
+
+        return LegacyExecutor
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+
+
+def _reset_discovery_for_tests() -> None:
+    """Re-arm entry point discovery so a test can observe it running again.
+
+    Discovery only. This does **not** restore the bundled executor: that is
+    registered by module import, which cannot be made to happen twice in one
+    process. A test that clears ``ExecutorRegistry`` must re-register
+    explicitly, as ``tests/test_legacy_executor_scaffold.py`` does.
+    """
+    global _cloud_executors
+    _cloud_executors = None
 
-# Discover and register cloud executors installed via entry points.
-# Each cloud executor class is decorated with @ExecutorRegistry.register,
-# so importing it (via ep.load()) is enough to register it.
-# If no cloud plugins are installed this returns an empty list.
-_cloud_executors = ExecutorPluginLoader.discover_executors()
 
-__all__ = ["LegacyExecutor"]
+__all__ = ["LegacyExecutor", "register_all"]

workers/executor/executors/plugins/loader.py

@@ -8,7 +8,7 @@
 
 - ``unstract.executor.executors``
     Executor classes that self-register via ``@ExecutorRegistry.register``.
-    Loaded eagerly at worker startup from ``executors/__init__.py``.
+    Loaded when ``executor.executors.register_all()`` runs at worker startup.
 """
 
 import logging

workers/executor/tasks.py

@@ -5,16 +5,9 @@
 ExecutionOrchestrator, and returns an ExecutionResult dict.
 """
 
-# Import the executor implementations so their ``@ExecutorRegistry.register``
-# decorators run before ``execute_extraction`` can be invoked. Coupling this to
-# the task module (not only the Celery ``executor/worker.py`` entrypoint) ensures
-# the registry is populated wherever the task is registered — in particular the PG
-# executor consumer, which bootstraps via the root-worker import that loads
-# ``executor/tasks.py`` (this module) but NOT ``executor/worker.py`` where this
-# import historically lived; without it the consumer hits "No executor
-# registered". Import is idempotent (module cached), so the Celery entrypoint
-# importing it again is harmless.
-import executor.executors  # noqa: E402, F401
+import logging
+
+from executor.executors import register_all
 from queue_backend import worker_task
 from shared.clients import UsageAPIClient
 from shared.enums.task_enums import TaskName
@@ -25,8 +18,30 @@
 from unstract.sdk1.execution.orchestrator import ExecutionOrchestrator
 from unstract.sdk1.execution.result import ExecutionResult
 
+# Suppress Celery trace logging of task return values.
+# The trace logger prints the full result dict on task success, which can
+# contain sensitive customer data (extracted text, summaries, etc.).
+#
+# This lives here, not in ``executor/worker.py``, because that module is not on
+# any deployed path: ``workers/worker.py`` exec-loads THIS file by path for both
+# the Celery and PG executor roles, and no launcher runs ``celery -A executor``.
+# It used to be reached only because ``executor/__init__.py`` eagerly imported
+# ``.worker``; making that lazy (UN-4136) would otherwise have silently
+# re-enabled result logging on every extraction.
+logging.getLogger("celery.app.trace").setLevel(logging.WARNING)
+
 logger = WorkerLogger.get_logger(__name__)
 
+# Populate ``ExecutorRegistry`` so the executor implementations are registered
+# before ``execute_extraction`` can be invoked. Coupling this to the task module
+# (not only the Celery ``executor/worker.py`` entrypoint) ensures the registry is
+# populated wherever the task is registered — in particular the PG executor
+# consumer, which bootstraps via the root-worker import that loads
+# ``executor/tasks.py`` (this module) but NOT ``executor/worker.py``; without it
+# the consumer hits "No executor registered". This is the only call site;
+# ``executor/worker.py`` reaches it by importing this module.
+register_all()
+
 _LLM_BEARING_OPS = frozenset(
     {
         "answer_prompt",

workers/executor/worker.py

@@ -4,7 +4,6 @@
 Routes execute_extraction tasks to registered executors.
 """
 
-import logging
 import os
 
 from queue_backend import worker_task
@@ -17,11 +16,6 @@
 logger = WorkerLogger.setup(WorkerType.EXECUTOR)
 app, config = WorkerBuilder.build_celery_app(WorkerType.EXECUTOR)
 
-# Suppress Celery trace logging of task return values.
-# The trace logger prints the full result dict on task success, which
-# can contain sensitive customer data (extracted text, summaries, etc.).
-logging.getLogger("celery.app.trace").setLevel(logging.WARNING)
-
 
 def check_executor_health():
     """Custom health check for executor worker."""
@@ -79,7 +73,8 @@ def healthcheck(self):
     }
 
 
-# Import tasks so shared_task definitions bind to this app.
-# Import executors to trigger @ExecutorRegistry.register at import time.
-import executor.executors  # noqa: E402, F401
+# Import tasks so shared_task definitions bind to this app. This also populates
+# ExecutorRegistry: executor/tasks.py calls register_all() at module scope, and
+# it is the site that does the work on every deployed path — workers/worker.py
+# exec-loads that file by path, and nothing launches `celery -A executor`.
 import executor.tasks  # noqa: E402, F401