This is an automated email from the ASF dual-hosted git repository.
ashb pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new a52be75711a Do not leak threads from InProcessExecutionAPI (#68840)
a52be75711a is described below
commit a52be75711a439cdb8c0627c20ce51df2214e796
Author: Nick Stenning <[email protected]>
AuthorDate: Mon Jun 22 17:39:04 2026 +0200
Do not leak threads from InProcessExecutionAPI (#68840)
Through `a2wsgi.ASGIMiddleware`, the `InProcessExecutionAPI` was leaking
a daemon thread per instance, used by `a2wsgi` to run an asyncio event
loop.
This commit moves management of that background thread into
`InProcessExecutionAPI`, and adds a finalizer so that the thread is
stopped when the execution API instance is garbage collected.
---
.../src/airflow/api_fastapi/execution_api/app.py | 40 ++++++++++++++++++----
.../unit/api_fastapi/execution_api/test_app.py | 26 ++++++++++++++
task-sdk/dev/generate_task_sdk_models.py | 2 ++
3 files changed, 62 insertions(+), 6 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 88c7d23fff2..12bbb0921a2 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/app.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/app.py
@@ -19,7 +19,9 @@ from __future__ import annotations
import asyncio
import json
+import threading
import time
+import weakref
from contextlib import AsyncExitStack
from functools import cached_property
from typing import TYPE_CHECKING, Any, cast
@@ -348,6 +350,23 @@ 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.
+def _shutdown_loop(
+ loop: asyncio.AbstractEventLoop,
+ thread: threading.Thread,
+ cm: AsyncExitStack,
+) -> None:
+ """Close the FastAPI lifespan and stop the background event loop +
thread."""
+ try:
+ asyncio.run_coroutine_threadsafe(cm.aclose(), loop).result(timeout=5)
+ except Exception:
+ logger.exception("Error while closing in-process execution API
lifespan")
+ loop.call_soon_threadsafe(loop.stop)
+ thread.join(timeout=5)
+
+
@attrs.define()
class InProcessExecutionAPI:
"""
@@ -358,7 +377,6 @@ class InProcessExecutionAPI:
"""
_app: FastAPI | None = None
- _cm: AsyncExitStack | None = None
@cached_property
def app(self):
@@ -393,20 +411,30 @@ class InProcessExecutionAPI:
@cached_property
def transport(self) -> httpx.WSGITransport:
- import asyncio
-
import httpx
from a2wsgi import ASGIMiddleware
- middleware = ASGIMiddleware(self.app)
+ # We choose to own the event loop + executor thread here so that we
can have explicit control over
+ # their lifecycle.
+ loop = asyncio.new_event_loop()
+ thread = threading.Thread(target=loop.run_forever,
name="InProcessExecutionAPI-loop", daemon=True)
+ thread.start()
+
+ middleware = ASGIMiddleware(self.app, loop=loop)
# https://github.com/abersheeran/a2wsgi/discussions/64
async def start_lifespan(cm: AsyncExitStack, app: FastAPI):
await cm.enter_async_context(app.router.lifespan_context(app))
- self._cm = AsyncExitStack()
+ cm = AsyncExitStack()
+
+ # Wait for lifespan startup to complete so callers see a ready app and
so the finalizer can
+ # 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)
- asyncio.run_coroutine_threadsafe(start_lifespan(self._cm, self.app),
middleware.loop)
return httpx.WSGITransport(app=middleware) # type: ignore[arg-type]
@cached_property
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 a5f371e565f..75e25bf0eba 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
@@ -17,6 +17,8 @@
from __future__ import annotations
import asyncio
+import gc
+import threading
from unittest import mock
from uuid import UUID
@@ -28,6 +30,7 @@ from fastapi.testclient import TestClient
from opentelemetry import context as otel_context, propagate as otel_propagate
from airflow.api_fastapi.execution_api.app import (
+ InProcessExecutionAPI,
_extract_w3c_trace_context,
create_task_execution_api_app,
)
@@ -134,6 +137,29 @@ 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.
+
+ Regression coverage for the a2wsgi background-thread leak.
+ """
+ before = {t for t in threading.enumerate() if t.name ==
"InProcessExecutionAPI-loop"}
+
+ api = InProcessExecutionAPI()
+ _ = api.transport # trigger loop + thread creation
+
+ 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.
+ del api
+ gc.collect()
+ thread.join(timeout=5)
+ assert not thread.is_alive()
+
+
class TestCorrelationIdMiddleware:
def test_correlation_id_echoed_in_response_headers(self, client):
"""Test that correlation-id from request is echoed back in response
headers."""
diff --git a/task-sdk/dev/generate_task_sdk_models.py
b/task-sdk/dev/generate_task_sdk_models.py
index bb19fffc449..74a1cb7f756 100644
--- a/task-sdk/dev/generate_task_sdk_models.py
+++ b/task-sdk/dev/generate_task_sdk_models.py
@@ -33,6 +33,8 @@ from datamodel_code_generator import (
from openapi_spec_validator import validate_spec
os.environ["_AIRFLOW__AS_LIBRARY"] = "1"
+# Set a placeholder secret to allow the in-process FastAPI app to run its
lifecycle hooks.
+os.environ.setdefault("AIRFLOW__API_AUTH__JWT_SECRET",
"task-sdk-model-generation")
AIRFLOW_ROOT_PATH = Path(__file__).parents[2].resolve()
AIRFLOW_TASK_SDK_ROOT_PATH = AIRFLOW_ROOT_PATH / "task-sdk"
