Migrate API endpoints from sync to async DB sessions
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 9h
- Merged PRs (30d)
- 472
Description
### Description
Incrementally convert Airflow's FastAPI routes from the synchronous `SessionDep` to the async `AsyncSessionDep`, so DB I/O no longer occupies an AnyIO threadpool worker and instead yields the event loop. The async infrastructure already ships in Airflow 3.x; adoption is what's missing.
This is an umbrella/tracking epic. Individual conversions land as their own small, parity-focused PRs.
### Use case/motivation
- The async SQLAlchemy plumbing already exists: `create_async_engine`, `async_sessionmaker`, `AsyncSessionDep`, `create_session_async`, `paginated_select_async`, the `sql_alchemy_conn_async` config key, and automatic async-driver derivation. None of this needs to be built — only adopted. Adoption inside the Execution API started at ~0 of ~49 routes.
- The relevant bottleneck on Airflow 3.x is the **Starlette/AnyIO threadpool inside the API server** (default 40 workers), not the Triggerer event loop. In 3.x the Triggerer and Workers reach the metadata DB through the Execution API over HTTP, not directly — so the original framing of PR #36504 ("unblock the Triggerer loop") no longer applies; the win is at the API server. See `dev/async_session_poc/PR36504_NOTES.md` and `RESULTS.md`.
- Measured A/B (single uvicorn worker): async wins cleanly at moderate concurrency (e.g. `/sleep` 500 ms: +23% RPS, −38% p99 at c=50) and inverts above the threadpool ceiling (c≥100) on a single worker. A follow-up multi-worker run (workers ∈ {1, 4}) with a multi-process client was inconclusive on a single 8-core box — the clean-regime async win holds (+2% to +24% RPS), but high-concurrency cells saturated and the 4-worker cells exhausted Postgres `max_connections` (see [`RESULTS_MULTIWORKER.md`](RESULTS_MULTIWORKER.md)). Notably, removing the client GIL did *not* remove the single-worker inversion, which locates it server-side (one event loop serialising CPU-bound request/response work), not in the client harness.
## Definition of done per endpoint
Each conversion PR must:
1. Convert `def` → `async def`, `SessionDep` → `AsyncSessionDep`, and `await` all `session.execute/scalar/scalars` calls. No `commit()` added in routes (the async dependency commits / rolls back, mirroring sync).
2. Preserve the contract exactly — same response model, query/path signature, and status codes. No Execution API version bump for a pure sync→async swap.
3. Pass the endpoint's existing tests **byte-identically** where they don't mock the session; update only session-mocking tests to patch `sqlalchemy.ext.asyncio.AsyncSession` with async interceptors.
4. Add the `reconfigure_async_db_engine` autouse fixture (calls `_configure_async_session()`) to any test class that exercises a converted route — the async engine binds its pool to the creating event loop, and the test harness uses a fresh loop per test. (Pattern: `TestWaitDagRun`, `TestTIHealthEndpoint`.)
5. Verify on **all three supported backends** — SQLite (aiosqlite), Postgres (asyncpg), MySQL (aiomysql) — not SQLite alone (`SELECT … FOR UPDATE` is a no-op there, and `rowcount` semantics differ per driver).
## Cross-cutting work (shared dependencies / risks)
- **Async secrets backends** — prerequisite for any route that resolves Variable/Connection *values*.
- **Multi-worker benchmark** — a first multi-worker (`AIRFLOW__API__WORKERS` ∈ {1, 4}) + multi-process-client run is done ([`RESULTS_MULTIWORKER.md`](RESULTS_MULTIWORKER.md)) but was inconclusive on a single shared-core box. A definitive crossover test still needs separate server/client hosts, per-worker pools sized so `pool_size × workers ≤ max_connections`, and a raised `max_connections`. Until a clean high-concurrency win is shown, conversions remain justified by parity, not by a measured throughput claim.
- **Async-engine disposal at teardown** — `dispose_orm` closes the async engine synchronously, logging a cosmetic `Event loop is closed` at session teardown on all async drivers (exit code unaffected). Worth tidying once, centrally.
- **Connection budget** — each converted route shifts load onto the async pool; a busy API server can hold both the sync and async pools. With multiple API-server workers, each worker holds its *own* async pool, so demand is roughly `pool_size × workers` — the multi-worker benchmark exhausted Postgres `max_connections=100` precisely because the per-worker pool was not sized for that multiplication. Size per-worker pools so `pool_size × workers` fits `max_connections`, and watch headroom as adoption grows.
## Guiding principles
- One endpoint (or a small, cohesive batch) per PR; parity over cleverness.
- Prefer read endpoints with comprehensive existing tests first; defer anything needing async secrets backends.
- Don't convert sibling routes opportunistically — keep diffs small and reviewable.
## Endpoints to convert
Execution API routes (`airflow-core/src/airflow/api_fastapi/execution_api/routes/`) currently on the synchronous `SessionDep`. Each box is one parity-focused conversion (small batches per file are fine). Checked = already async.
**`task_instances.py`**
- [ ] `PATCH /task-instances/{task_instance_id}/run` — `ti_run`
- [ ] `PATCH /task-instances/{task_instance_id}/state` — `ti_update_state`
- [ ] `PATCH /task-instances/{task_instance_id}/skip-downstream` — `ti_skip_downstream`
- [ ] `PUT /task-instances/{task_instance_id}/heartbeat` — `ti_heartbeat` — https://github.com/apache/airflow/pull/67800
- [ ] `PUT /task-instances/{task_instance_id}/rtif` — `ti_put_rtif`
- [ ] `PATCH /task-instances/{task_instance_id}/rendered-map-index` — `ti_patch_rendered_map_index`
- [ ] `GET /task-instances/{task_instance_id}/previous-successful-dagrun` — `get_previous_successful_dagrun`
- [ ] `GET /task-instances/count` — `get_task_instance_count`
- [ ] `GET /task-instances/previous/{dag_id}/{task_id}` — `get_previous_task_instance`
- [ ] `GET /task-instances/states` — `get_task_instance_states`
- [ ] `GET /task-instances/breadcrumbs` — `get_task_instance_breadcrumbs`
- [ ] `GET /task-instances/{task_instance_id}/validate-inlets-and-outlets` — `validate_inlets_and_outlets`
**`dag_runs.py`**
- [ ] `GET /dag-runs/{dag_id}/previous` — `get_previous_dagrun_compat`
- [ ] `GET /dag-runs/{dag_id}/{run_id}` — `get_dag_run`
- [ ] `POST /dag-runs/{dag_id}/{run_id}` (trigger) — `trigger_dag_run`
- [ ] `POST /dag-runs/{dag_id}/{run_id}/clear` — `clear_dag_run`
- [ ] `GET /dag-runs/{dag_id}/{run_id}/state` — `get_dagrun_state`
- [ ] `GET /dag-runs/count` — `get_dr_count`
- [ ] `GET /dag-runs/previous` — `get_previous_dagrun`
**`xcoms.py`**
- [ ] `GET /xcoms/{dag_id}/{run_id}/{task_id}/{key}/item/{offset}` — `get_mapped_xcom_by_index`
- [ ] `GET /xcoms/{dag_id}/{run_id}/{task_id}/{key}/slice` — `get_mapped_xcom_by_slice`
- [ ] `HEAD /xcoms/{dag_id}/{run_id}/{task_id}/{key}` — `head_xcom`
- [ ] `GET /xcoms/{dag_id}/{run_id}/{task_id}/{key}` — `get_xcom`
- [ ] `POST /xcoms/{dag_id}/{run_id}/{task_id}/{key}` — `set_xcom`
- [ ] `DELETE /xcoms/{dag_id}/{run_id}/{task_id}/{key}` — `delete_xcom`
**`asset_state.py`**
- [ ] `GET /state/asset/by-name/value` — `get_asset_state_by_name`
- [ ] `PUT /state/asset/by-name/value` — `set_asset_state_by_name`
- [ ] `DELETE /state/asset/by-name/value` — `delete_asset_state_by_name`
- [ ] `DELETE /state/asset/by-name/clear` — `clear_asset_state_by_name`
- [ ] `GET /state/asset/by-uri/value` — `get_asset_state_by_uri`
- [ ] `PUT /state/asset/by-uri/value` — `set_asset_state_by_uri`
- [ ] `DELETE /state/asset/by-uri/value` — `delete_asset_state_by_uri`
- [ ] `DELETE /state/asset/by-uri/clear` — `clear_asset_state_by_uri`
**`asset_events.py`**
- [ ] `GET /asset-events/by-asset` — `get_asset_event_by_asset_name_uri`
- [ ] `GET /asset-events/by-asset-alias` — `get_asset_event_by_asset_alias`
**`assets.py`**
- [ ] `GET /assets/by-name` — `get_asset_by_name`
- [ ] `GET /assets/by-uri` — `get_asset_by_uri`
- [ ] `GET /assets/by-alias` — `get_assets_by_alias`
**`task_state.py`**
- [ ] `GET /state/ti/{task_instance_id}/{key}` — `get_task_state`
- [ ] `PUT /state/ti/{task_instance_id}/{key}` — `set_task_state`
- [ ] `DELETE /state/ti/{task_instance_id}/{key}` — `delete_task_state`
- [ ] `DELETE /state/ti/{task_instance_id}` — `clear_task_state`
**`hitl.py`**
- [ ] `POST /hitlDetails/{task_instance_id}` — `upsert_hitl_detail`
- [ ] `PATCH /hitlDetails/{task_instance_id}` — `update_hitl_detail`
- [ ] `GET /hitlDetails/{task_instance_id}` — `get_hitl_detail`
**`dags.py`**
- [ ] `GET /dags/{dag_id}` — `get_dag`
**`task_reschedules.py`**
- [ ] `GET /task-reschedules/{task_instance_id}/start_date` — `get_start_date`
**`variables.py`**
- [ ] `GET /variables/keys` — `get_variable_keys` — draft in c2d8c2c0
### Deferred: routes blocked on async secrets backends
These resolve Variable/Connection *values* through `Variable.get/set/delete` and the secrets backend rather than a route-level `SessionDep`, so they are not simple sync→async session swaps. They convert only once async secrets backends land (see Cross-cutting work above), and are tracked here for completeness:
- [ ] `GET /variables/{variable_key}` — `get_variable`
- [ ] `PUT /variables/{variable_key}` — `put_variable`
- [ ] `DELETE /variables/{variable_key}` — `delete_variable`
- [ ] `GET /connections/{connection_id}` — `get_connection`
## References
- Benchmark code / experiments at: https://github.com/apache/airflow/compare/main...Dev-iL:airflow:2605/async_sqla_poc
(see `dev/async_session_poc/PR36504_NOTES.md`, `RESULTS.md`, `results.csv`, `diagnostics/pg_stat_activity.log`)
### Related issues
- #36504
### Are you willing to submit a PR?
- [x] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
Contributor guide
Assessment
This issue has not been assessed yet.