jason810496 opened a new issue, #76425: URL: https://github.com/apache/airflow/issues/76425
### Body ### Context - An `async def` FastAPI dependency that runs a sync query waits for a pooled connection on the event loop. When the pool is full, the loop blocks, and the requests holding connections can't release them, because their teardown needs the loop too. That stalled 3.4.0rc2 under concurrent writes (#76417, found in #75865 ([comment](https://github.com/apache/airflow/issues/75865#issuecomment-6622866800))). - #76416 fixes the 3.4.0 regression in `log_action`. #76424 adds tests that fail on sync DB work on the event loop, with the known offenders listed as strict xfails in `generated/`. - The items below are the same pattern. They were already in 3.3.2 and are off the common path, so they are 3.4.1 follow-ups, not RC blockers. Under load each one still wedges the whole server: a single call per second is enough once the pool is busy. ### Follow-up items - [ ] `requires_access_backfill` in `core_api/security.py`: looks up `Backfill.dag_id` on `/backfills/{backfill_id}` and its `/dag_runs`, `/pause`, `/unpause` and `/cancel` routes. `DagModel.get_team_name` also runs on the loop when its cache misses. - [ ] `requires_access_event_log` in `core_api/security.py`: looks up the `Log` row on `GET /eventLogs/{event_log_id}`. - [ ] `_collect_teams_to_check` in `core_api/security.py` (only with `[core] multi_team = True`): calls the sync `Pool`, `Connection` or `Variable` `get_team_name` on detail, update and delete routes. It also calls `Team.get_name_if_exists` when a create or update body sets `team_name`. - [ ] `BaseAuthManager.get_user_from_token`: replace `run_in_threadpool(RevokedToken.is_revoked, ...)` with an async query. It's listed in `generated/known_api_server_threadpool_offloads.txt`. - [ ] `GET /api/v2/dags`: `get_dags` passes the lazy `session.scalars(...)` into `Iterable[DAGResponse]`, so the rows and the `dag_tag` query are read while FastAPI serializes the response on the loop. Materialize it with `.all()`, and check the other list routes whose collection field is `Iterable[...]`. #74247 fixed the same shape for Dag details and pools. - [ ] FAB provider: `FabAuthManager.deserialize_user` runs a sync user query from `get_user_from_token` on the loop each time its per-user cache (`[fab] cache_ttl`) expires. - [ ] Coverage (`test_event_loop_blocking.py`): the sweep sends `"1"` for every path parameter and `{}` as the body against an empty DB. Routes that reject that input early, or that only query when rows exist (such as the `GET /dags` case), are not covered. The mounted `/execution` app isn't covered either, because the sweep's UI token is rejected there before any route dependency runs. ### How to fix 1. Run the lookup on the async engine instead of the sync session. Take `AsyncSessionDep`, or use `async with create_session_async() as session:`, and `await session.scalar(...)`, as #76416 does for `log_action`. Add async variants of the sync `@provide_session` helpers where needed. Don't wrap the call in `run_in_threadpool`: `TestNoThreadpoolOffload` fails on any new offload. 2. Delete the fixed routes' lines from `generated/known_sync_db_on_event_loop.txt`, or from `generated/known_api_server_threadpool_offloads.txt` for `is_revoked`. Each line is a strict xfail, so the run fails until the line is removed. ### Committer - [x] I acknowledge that I am a maintainer/committer of the Apache Airflow project. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
