seanmuth opened a new issue, #66358:
URL: https://github.com/apache/airflow/issues/66358
## Summary
In Airflow 3.2, deferrable triggers that fetch XComs (or Variables /
Connections) from inside `run()` are funneled through a single serialized
supervisor IPC channel. Under moderate trigger fan-out (observed with
`astronomer-cosmos==1.14.1` + `dbt-snowflake==1.11.4`, where each
`WatcherTrigger` calls `XCom.get_one` per poke), individual triggers exceed
their `defer(timeout=...)` and surface as `TaskDeferralTimeout`. The same
workload runs cleanly on 3.1.
## Root cause
XCom access from triggers (added in 683cae4a7d, "Enable Comms in the
TriggerRunner to make Connections, Variables and XComs accessible to Triggers")
routes through `SUPERVISOR_COMMS.send`. On the trigger side this is
`TriggerCommsDecoder.send`, which acquires a process-global `threading.Lock`
and calls `async_to_sync(self.asend)` on the triggerer's main event loop:
```python
# airflow-core/src/airflow/jobs/triggerer_job_runner.py:906-910
def send(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
from asgiref.sync import async_to_sync
with self._thread_lock:
return async_to_sync(self.asend)(msg)
```
On the supervisor side, `_handle_request` is dispatched from a single sync
`run_once` loop and services the request via an in-process Execution API client
doing a synchronous DB query:
- `airflow-core/src/airflow/jobs/triggerer_job_runner.py:457-508`
(`_handle_request` dispatch, including the `GetXCom` branch that calls
`self.client.xcoms.get(...)`)
- `airflow-core/src/airflow/jobs/triggerer_job_runner.py:581-611`
(single-threaded `run`/`run_once` loop)
Net effect: every trigger XCom fetch in the process serializes through two
single-threaded chokepoints — the `_thread_lock` on the trigger side and the
supervisor's `run_once` loop on the other side. `sync_to_async(XCom.get_one)`
from user code (e.g. `cosmos/operators/_watcher/triggerer.py:78`) does not help
— the asgiref threadpool just queues behind the lock.
3.1 had no such mechanism; XCom access from triggers went directly to the DB
via SQLAlchemy and parallelized across the connection pool.
## Reproduction
DAG with N (>~20) deferrable tasks using cosmos `WatcherTrigger`, or any
trigger that calls `XCom.get_one`/`get_value` from inside `run()` on a short
poke interval. Default triggerer config. Snowflake or any backend where the
supervisor's DB query is non-trivial amplifies the effect.
## Suggested fixes
1. Expose an awaitable `XCom.aget_one` (and Variables/Connections
equivalents) that calls `SUPERVISOR_COMMS.asend` directly, bypassing
`async_to_sync`. Trigger authors should `await` it.
2. Replace the global `_thread_lock` with a per-request mechanism — request
IDs are already present in the protocol, so the writer queue does not need to
be serialized end-to-end.
3. Allow the supervisor to service `GetXCom`/`GetVariable`/`GetConnection`
concurrently — these are read-only and safe to parallelize. A worker pool or
async DB session would let multiple requests proceed simultaneously.
Cosmos and similar providers can then migrate to the async API and the
regression is removed.
## Environment
- Airflow 3.2.x (regressed from 3.1)
- `astronomer-cosmos==1.14.1`, `dbt-snowflake==1.11.4`
- Snowflake metadata DB backend
---
Drafted-by: Claude Code (Opus 4.7); reviewed by @seanmuth before posting
--
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]
