Dev-iL opened a new pull request, #75396: URL: https://github.com/apache/airflow/pull/75396
related: https://github.com/apache/airflow/issues/67799, https://github.com/apache/airflow/pull/67800, https://github.com/apache/airflow/pull/73405 Convert the six Execution API XCom routes (`GET .../item/{offset}`, `GET .../slice`, `HEAD`, `GET`, `POST` and `DELETE` on `/execution/xcoms/{dag_id}/{run_id}/{task_id}/{key}`) to `AsyncSessionDep`, so database waits release threadpool worker capacity. The HTTP contract is unchanged: paths, parameters, status codes and error bodies, team-scoped access, trigger-identity checks, loop and mapped-producer resolution, slice and offset semantics, and every API version's behaviour. The OpenAPI document is byte-identical to `main` for all 11 Execution API versions. The session dependency still owns commit and rollback before the response is sent. > [!IMPORTANT] > **Each XCom request now uses two database connections instead of one: one from the sync pool and one from the async pool.** > Producer resolution runs in a worker thread with its own sync session, so a Dag deserialized on a DagBag cache miss doesn't block the event loop. The route itself uses an `AsyncSession`. | Request | Connections checked out | Held at the same time | | --- | --- | --- | | `GET` single value with an execution token | 1 async + 1 sync | Yes: the owner check queries first | | Any route with the trigger header, or with `[core] multi_team` on | 1 async + 1 sync | Yes: the access check queries first | | All other `GET`, `HEAD`, item and slice reads, `POST`, `DELETE` | 1 sync, then 1 async | No | <details><summary><h2>Details & Benchmarks</h2></summary> <p> ### What runs where - `has_xcom_access` (the router dependency), `_resolve_trigger_token` and the handlers share the route's `AsyncSession`. A test that counts distinct sessions per request confirms the dependencies did not get their own. - Producer resolution (`_build_xcom_read`, `_find_writer_id`) runs in `run_in_threadpool` and opens its own sync session. It goes through `TaskCoordinateResolver`, which can deserialize a serialized Dag on a DagBag cache miss. Under `AsyncSession.run_sync`, that CPU work would run on the event loop. This is a compatibility bridge: it uses worker-thread capacity as before. The two helpers return a statement or an id, never ORM objects. - `XComModel.set_for_attempt` and `delete_for_attempts` stay single implementations shared with sync callers, called through `run_sync` on the route's session. - `get_xcom` and the item route select only the `value` column instead of hydrating the XCom projection, and they test the fetched row so that a stored `NULL` value still returns 200 rather than 404. The slice reads its values with `scalars()`. - The three read routes return a `JSONResponse` built in `run_in_threadpool`. The values come from a JSON column, so this skips FastAPI's re-validation of every element as `JsonValue`, which an `async def` route would otherwise do on the event loop. `response_model` stays on each decorator, which keeps the schema. No Cadwyn version change migrates these responses. Cadwyn still decodes and re-encodes a returned `JSONResponse` body on the loop (zmievsa/cadwyn#51). Both pools are sized per API server process by `[database] sql_alchemy_pool_size` (default 5) plus `sql_alchemy_max_overflow` (default 10). In the worst case a process holds 15 + 15 metadata DB connections, where it used to hold at most 15 sync connections for these routes. Budget that across workers and replicas, and with PgBouncer if you use it. Pool waits cannot deadlock: a request holding an async connection may wait for a sync one, but the sync-only lookup never waits for the async pool. For `POST` and `DELETE`, the target task instance is resolved in the lookup's transaction and written in the route's. No row lock is taken on that path, so none is released early. ## Benchmarks Measured with a disposable harness that drives the real HTTP stack while a 1 ms ticker on the server loop records each sleep. There were 11 sequential requests per scenario, the first dropped, and each revision ran twice. Cells are median / maximum stall in ms for run 1; run 2. Base is `main` at `8cc830d4ef`; Python 3.14.8 and SQLAlchemy 2.0.54 in Breeze. | Scenario | Backend | Base | This PR | | --- | --- | --- | --- | | `GET`, one 3.9 MB value | PostgreSQL | 977 / 1085; 965 / 1028 | 165 / 553; 164 / 556 | | `GET`, one 3.9 MB value | SQLite | 930 / 943; 936 / 959 | 151 / 156; 157 / 175 | | Slice of 5,000 values | PostgreSQL | 53 / 68; 56 / 65 | 36 / 51; 43 / 70 | | Slice of 5,000 values | SQLite | 58 / 398; 35 / 450 | 50 / 56; 50 / 56 | | Loop-task read, Dag cache cleared before each request | PostgreSQL | 8.8 / 14.7; 11.0 / 17.6 | 10.6 / 14.2; 11.7 / 21.6 | | Loop-task read, Dag cache cleared before each request | SQLite | 10.6 / 13.2; 9.6 / 18.1 | 10.7 / 14.2; 11.2 / 13.7 | | Loop-task read, warm cache | PostgreSQL | 6.3 / 6.7; 6.3 / 7.2 | 5.4 / 7.9; 4.6 / 5.7 | | Loop-task read, warm cache | SQLite | 6.3 / 6.4; 6.2 / 8.6 | 10.2 / 11.7; 10.9 / 13.8 | A plain `AsyncSession.run_sync` conversion, measured first, regressed the large value (1156–1847 ms median) and the cold-cache loop read (71–110 ms). The loop-thread profile put 1.1 s of the 2.1 s large-value request in pydantic's `JsonValue` validation. The remaining cost is the warm loop read on SQLite, about 4 ms of median stall: the hop to a worker thread, which shows only on the SQLite driver. These are small local samples of event-loop availability, not throughput claims. The harness is not part of this PR. ## Tests - Existing HTTP assertions are unchanged. The access-denied fixture now awaits the async check with its own async session. - The trigger test that counted one session per request now expects two per request (12 for its 6 requests), the route's and the lookup's. It still fails if a dependency opens a third. - New: `test_xcom_read_returns_a_stored_null_without_loading_xcom_entities` (`get` and `item`) fails on `main`, where each read loads an `XComModel`. - New: `test_xcom_producer_lookups_and_rendering_run_off_the_event_loop` (six routes) asserts the offloaded helpers run without an event loop in their thread. It also passes against the synchronous handlers on `main`, which already ran in a worker thread. Its failing control is this PR with the helpers called directly on the loop: all six cases fail. Validation through Breeze, Python 3.14.8: | Backend | Async driver | Result | | --- | --- | --- | | SQLite | `sqlite+aiosqlite` | 170 passed | | PostgreSQL | `postgresql+psycopg_async` | 170 passed | | MySQL | `mysql+aiomysql` | 170 passed | Command for each backend: ```bash breeze run --python 3.14 --backend <backend> --answer y pytest \ airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py \ airflow-core/tests/unit/api_fastapi/execution_api/versions/v2025_04_28/test_xcom.py \ -q -p no:cacheprovider --with-db-init ``` The suites that reach these routes through other entry points (in-process API, triggerer job, token scope boundaries, security, Dag processor, `v2026_10_30` task instances) passed on SQLite: 575 passed, 1 skipped. Fast and manual prek checks passed, including Ruff, mypy and the API generation checks; no generated files changed. Of the suites `breeze verify` lists, `API Always` (DB and non-DB), the `common.compat` and `fab` non-DB provider tests, and the docs build passed. Three failures there are local and reproduce on `main`: `test_providers_manager` and the `fab` DB tests crash with `Illegal instruction` (the development host lacks AVX2), and the two `TestArgBindingsFieldBackwardCompat` cases that errored in the parallel run pass on their own, on `main` and on this branch. </p> </details> --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes — Claude Code (Claude Opus 5.5) Generated-by: Claude Code (Claude Opus 5.5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) -- 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]
