From 48b2427c35690d67365ca3a1239d6741dec66d78 Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Mon, 22 Jun 2026 15:02:51 -0400 Subject: [PATCH 1/2] Fix in-process Execution API loop stopped while its transport is in use The in-process Execution API owns a background asyncio event loop + daemon thread. Their cleanup was registered with `weakref.finalize` keyed on the `InProcessExecutionAPI` instance. But callers build a sync `Client` from `InProcessExecutionAPI().transport` and discard the factory object, so the instance was garbage-collected while the transport was still in use. The finalizer then stopped the loop, and every subsequent request hung on the dead loop -- surfacing as `Timeout` failures across the Dag-processor and triggerer in-process API tests on main. Tie the loop + thread + lifespan cleanup to the lifetime of the WSGI transport (what callers actually retain) instead of the factory instance. This keeps the original thread-leak fix intact -- the loop is still stopped and the thread joined once nothing can use the transport -- while no longer tearing it down prematurely. --- .../airflow/api_fastapi/execution_api/app.py | 18 ++++++++++++------ .../api_fastapi/execution_api/test_app.py | 19 +++++++++++++------ 2 files changed, 25 insertions(+), 12 deletions(-) diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/app.py b/airflow-core/src/airflow/api_fastapi/execution_api/app.py index 12bbb0921a25f..7a2b4cd5c67ef 100644 --- a/airflow-core/src/airflow/api_fastapi/execution_api/app.py +++ b/airflow-core/src/airflow/api_fastapi/execution_api/app.py @@ -350,9 +350,9 @@ def get_extra_schemas() -> dict[str, dict]: } -# Note: _shutdown_loop is used as a finalizer for :class:`InProcessExecutionAPI`. As such, its arguments must -# not directly or indirectly reference the instance itself, as this will prevent the instance from being -# garbage collected. +# Note: _shutdown_loop is used as a finalizer for the WSGI transport returned by +# :meth:`InProcessExecutionAPI.transport`. As such, its arguments must not directly or indirectly reference +# that transport, as this would prevent the transport from being garbage collected. def _shutdown_loop( loop: asyncio.AbstractEventLoop, thread: threading.Thread, @@ -432,10 +432,16 @@ async def start_lifespan(cm: AsyncExitStack, app: FastAPI): # safely aclose() a context whose __aenter__ has actually run. asyncio.run_coroutine_threadsafe(start_lifespan(cm, self.app), loop).result() - # Stop the loop + thread and unwind the lifespan when this instance is garbage collected. - weakref.finalize(self, _shutdown_loop, loop, thread, cm) + transport = httpx.WSGITransport(app=middleware) # type: ignore[arg-type] - return httpx.WSGITransport(app=middleware) # type: ignore[arg-type] + # Stop the loop + thread and unwind the lifespan when the *transport* is garbage collected, not + # this InProcessExecutionAPI instance. Callers commonly build a Client from ``.transport`` and + # drop the factory object (e.g. ``client = Client(transport=InProcessExecutionAPI().transport)``); + # finalizing on ``self`` would stop the loop while the transport is still in use, so every later + # request would hang on the dead loop. + weakref.finalize(transport, _shutdown_loop, loop, thread, cm) + + return transport @cached_property def atransport(self) -> httpx.ASGITransport: diff --git a/airflow-core/tests/unit/api_fastapi/execution_api/test_app.py b/airflow-core/tests/unit/api_fastapi/execution_api/test_app.py index 75e25bf0eba6b..32880773ade2e 100644 --- a/airflow-core/tests/unit/api_fastapi/execution_api/test_app.py +++ b/airflow-core/tests/unit/api_fastapi/execution_api/test_app.py @@ -137,25 +137,32 @@ def test_routes_with_task_instance_id_param_enforce_ti_self(client): ) -def test_in_process_execution_api_teardown(): - """Accessing .transport spins up a daemon thread; dropping the instance must stop it via finalize. +def test_in_process_execution_api_transport_lifecycle(): + """The background loop + thread lifecycle is tied to the ``.transport``, not the factory instance. - Regression coverage for the a2wsgi background-thread leak. + Callers build a sync ``Client`` from ``InProcessExecutionAPI().transport`` and drop the factory + object. Dropping the instance must NOT stop the loop while the transport is still held -- doing so + left every later request hanging on a stopped loop. Dropping the transport must stop the loop and + join the daemon thread (the a2wsgi background-thread leak guard). """ before = {t for t in threading.enumerate() if t.name == "InProcessExecutionAPI-loop"} api = InProcessExecutionAPI() - _ = api.transport # trigger loop + thread creation + transport = api.transport # triggers loop + thread creation; the transport is what callers keep new_threads = {t for t in threading.enumerate() if t.name == "InProcessExecutionAPI-loop"} - before assert len(new_threads) == 1 thread = new_threads.pop() assert thread.is_alive() - # Drop the only strong reference; the weakref.finalize registered in .transport must stop - # the loop and join the daemon thread once the instance is collected. + # Dropping only the factory instance must leave the loop running for the still-live transport. del api gc.collect() + assert thread.is_alive() + + # Dropping the transport runs the weakref.finalize: loop stopped, daemon thread joined (no leak). + del transport + gc.collect() thread.join(timeout=5) assert not thread.is_alive() From e33945f54e0cb8bdc0c03edbf33dd7c861906536 Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Mon, 22 Jun 2026 16:29:06 -0400 Subject: [PATCH 2/2] Apply suggestion from @potiuk --- airflow-core/src/airflow/api_fastapi/execution_api/app.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/app.py b/airflow-core/src/airflow/api_fastapi/execution_api/app.py index 7a2b4cd5c67ef..7823e8ce3401b 100644 --- a/airflow-core/src/airflow/api_fastapi/execution_api/app.py +++ b/airflow-core/src/airflow/api_fastapi/execution_api/app.py @@ -351,7 +351,7 @@ def get_extra_schemas() -> dict[str, dict]: # Note: _shutdown_loop is used as a finalizer for the WSGI transport returned by -# :meth:`InProcessExecutionAPI.transport`. As such, its arguments must not directly or indirectly reference +`InProcessExecutionAPI.transport`. As such, its arguments must not directly or indirectly reference # that transport, as this would prevent the transport from being garbage collected. def _shutdown_loop( loop: asyncio.AbstractEventLoop,