Cache InProcessExecutionAPI in dag.test() to avoid per-task FastAPI app creation

[kaxil]

Summary

dag.test() creates a new InProcessExecutionAPI for every task execution. Each instance spins up a full FastAPI app with Cadwyn versioning, plus an a2wsgi ASGI-to-WSGI bridge that starts its own event loop thread. The dominant cost is the cold event loop thread startup and DB connection establishment per HTTP request.

Add @functools.lru_cache(maxsize=1) to in_process_api_server() so the FastAPI app and warm event loop are reused across all tasks within a dag.test() call. DagFileProcessorManager already uses this caching pattern (manager.py:269).

Also clean up stale dependency_overrides when _api_client(dag=None) is called (e.g. from run_trigger_in_process), and add a cache_clear fixture + caching contract test.

Benchmarks (breeze, SQLite)

Per-task overhead (profiled, 10-task noop DAG):

• Uncached: 1.25s/task -- Cached: 0.20s/task -- 6x within a single dag.test() call

Test suite improvement (avg of 3 runs, cache cleared between tests):

Test suite	Without cache	With cache	Improvement
test_dag.py (4 dag_test tests)	27.90s	21.58s	23% faster
test_mappedoperator.py (3 dag tests)	26.64s	17.04s	36% faster

The test suite improvement is smaller because fixture setup/teardown and cache_clear() between tests dominate. For downstream projects like Cosmos that run multi-task DAGs with a session-scoped cache, the per-task speedup applies directly -- consistent with their reported 47min->30min improvement.

Bottleneck breakdown (5-task profiled run, uncached):

• _thread.lock.acquire (a2wsgi sync bridge): 91%
• _api_client() (FastAPI+Cadwyn creation): 7%
• Other (task execution, DB, logging): 2%

Why this is safe

• dependency_overrides cleanup: _api_client(dag=None) now pops stale overrides from the cached instance
• Single-threaded context: dag.test() runs tasks sequentially
• No mocks to bypass: No test mocks in_process_api_server()
• cache_clear fixture: Autouse in airflow-core/tests/conftest.py prevents cross-test state leakage
• 303 tests pass: test_supervisor (123), test_dagrun (154), test_dag_command (12), test_dag (4), test_mappedoperator (3), test_xcom_arg (10)

Downstream impact

Cosmos CI tests (astronomer/astronomer-cosmos#2547) worked around this with a session-scoped pytest fixture that monkeypatches in_process_api_server(). This fix makes that workaround unnecessary.

[Copilot]

Pull request overview

This PR improves dag.test() performance by reusing a single InProcessExecutionAPI instance instead of creating a new FastAPI/a2wsgi stack per task execution.

Changes:

• Cache in_process_api_server() using functools.lru_cache(maxsize=1) to reuse the in-process Execution API server across task executions.
• Ensure stale dependency_overrides for dag_bag_from_app are removed when _api_client(dag=None) is used.
• Add tests/fixtures to validate and manage the caching contract (cache_clear() usage).

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.

File	Description
task-sdk/src/airflow/sdk/execution_time/supervisor.py	Caches in_process_api_server() and clears dag_bag_from_app dependency override when no DAG is provided.
task-sdk/tests/task_sdk/execution_time/test_supervisor.py	Adds a unit test asserting the caching/clearing behavior of in_process_api_server().
airflow-core/tests/conftest.py	Adds an autouse fixture to clear the cached in-process API server after each test to prevent state leakage.

[pankajkoti]

Thanks for optimising this piece!

[kaxil]

Static check failure is unrelated