Summary

In Airflow 3.2, deferrable triggers that fetch XComs (or Variables / Connections) from inside run() are funneled through a process-global lock around the 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 — what changed between 3.1 and 3.2

The supervisor-comms XCom path itself was added in 3.0.0 (683cae4a7d, "Enable Comms in the TriggerRunner to make Connections, Variables and XComs accessible to Triggers", #48239) and was present in 3.1.

What changed for 3.2 is #64882 (f102ad7f66, "Lock TriggerCommsDecoder sync req-res cycle"), which wrapped TriggerCommsDecoder.send and _read_frame in a process-global threading.Lock:

 def send(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
     from asgiref.sync import async_to_sync
-    return async_to_sync(self.asend)(msg)
+    with self._thread_lock:
+        return async_to_sync(self.asend)(msg)

Current code: airflow-core/src/airflow/jobs/triggerer_job_runner.py:900-910.

The PR fixed a real correctness issue — without the lock, multiple threads from sync_to_async's threadpool could interleave frames on the socket. But the lock now serializes every trigger's XCom/Variable/Connection request through one mutex, in a process where the supervisor side is also single-threaded (run_once loop at lines 581-611 dispatches _handle_request at lines 457-508 one message at a time, with GetXCom calling self.client.xcoms.get(...) synchronously against the in-process Execution API).

Net effect after 3.2: every trigger XCom fetch in the process is strictly serial — bounded by lock acquire → asend → supervisor DB query → reply → lock release. Any user code using sync_to_async(XCom.get_one) (e.g. cosmos/operators/_watcher/triggerer.py:78) gets no parallelism from the asgiref threadpool because the threads queue on the lock.

In 3.1 the same code path was racy but pipelined; the user's workload either didn't hit the race or hit it benignly, and benefited from the overlap. In 3.2 the correctness fix collapses throughput.

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.

Suggested fixes

The lock was added for a real reason (frame interleaving) and shouldn't simply be removed. Options:

1. Expose an awaitable API (XCom.aget_one, Variable.aget, Connection.aget) that calls SUPERVISOR_COMMS.asend directly. The IPC protocol already uses request IDs, so concurrent in-flight requests on the loop would not interleave at the framing level — only the writer/reader needs to be serialized, not the full req-res cycle. Trigger authors await instead of going through sync_to_async.

2. Narrow the lock to just the framing (writer mutex around the bytes write, reader dispatch by request ID) so that req-res cycles can overlap. This is what the request-ID design seems to anticipate.

3. Parallelize the supervisor side for read-only handlers (GetXCom, GetVariable, GetConnection) — these are safe to service concurrently and would let multiple in-flight requests proceed.

(1) by itself unblocks providers like Cosmos to migrate to the async API and removes the user-visible regression without touching the existing sync path.

Environment

• Airflow 3.2.x (regressed from 3.1)
• astronomer-cosmos==1.14.1, dbt-snowflake==1.11.4

Drafted-by: Claude Code (Opus 4.7); reviewed by @seanmuth before posting