This is an automated email from the ASF dual-hosted git repository.
jedcunningham pushed a commit to branch v3-2-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-2-test by this push:
new 46eca447a5b Fix triggerer crash when multiple triggers call sync SDK
methods concurrently (#66412) (#66446)
46eca447a5b is described below
commit 46eca447a5bd35c0014ec7ec68ce7d2a43b4baa9
Author: Jed Cunningham <[email protected]>
AuthorDate: Tue May 5 19:18:42 2026 -0600
Fix triggerer crash when multiple triggers call sync SDK methods
concurrently (#66412) (#66446)
TriggerCommsDecoder used an asyncio.Lock to serialise concurrent requests,
then read the response while holding the lock. When a trigger called a
sync SDK method wrapped with sync_to_async(), that method ran in a worker
thread and internally used async_to_sync() to make async calls, spinning
up a second event loop in that thread. The two event loops then raced to
read from the same asyncio.StreamReader, producing "Response read out of
order!" and crashing the TriggerRunner subprocess. PR #64882 attempted a
threading.Lock fix but introduced a new deadlock where the triggerer
heartbeat normally while processing zero triggers.
Replace the lock-based serial approach with response multiplexing:
* Each asend() registers an asyncio.Future keyed by frame.id in a
_pending dict. A single _reader_loop background task reads frames
one at a time and dispatches each response to the right waiter.
No lock, no ordering assumption.
* sync send() from worker threads schedules asend() on the main event
loop via run_coroutine_threadsafe() and blocks the calling thread.
This eliminates competing event loops entirely.
* sync send() from the event loop thread itself (e.g. a trigger calling
a sync SDK method directly from async def run() via greenback) would
deadlock with run_coroutine_threadsafe(...).result() because .result()
blocks the thread the loop runs on. Detected via thread-ID comparison;
greenback.await_() is used instead to teleport the coroutine back into
the running loop.
* asend() called from a foreign event loop (async_to_sync inside a
sync_to_async-wrapped trigger) detects the loop mismatch and bridges
back to the main loop via run_coroutine_threadsafe + wrap_future so
the foreign loop can still await the result. This is the root-cause
code path, covered directly by test_foreign_loop_path.
* init_comms() reads the initial StartTriggerer frame before starting
the reader loop so the loop never races with initialisation.
Add a subprocess event-loop watchdog: TriggerRunnerSupervisor stamps
_last_runner_comms on every message received from the subprocess. If the
subprocess goes silent for longer than the new [triggerer]
runner_health_check_threshold (default 30 s), heartbeat() skips the DB
update so the scheduler sees the triggerer as unhealthy and reassigns its
triggers. This detects a deadlocked or hung event loop that the
process-alive check alone cannot catch.
(cherry picked from commit 099fd2b03f3a4e642f75e748fbce2f3d99ffbba8)
Co-authored-by: Ash Berlin-Taylor <[email protected]>
---
airflow-core/newsfragments/66412.significant.rst | 26 +++
.../src/airflow/config_templates/config.yml | 11 +
.../src/airflow/jobs/triggerer_job_runner.py | 127 ++++++++---
airflow-core/tests/unit/jobs/test_triggerer_job.py | 231 ++++++++++++++++++++-
.../sdk/execution_time/secrets/execution_api.py | 40 ----
.../tests/task_sdk/execution_time/test_secrets.py | 101 ---------
6 files changed, 363 insertions(+), 173 deletions(-)
diff --git a/airflow-core/newsfragments/66412.significant.rst
b/airflow-core/newsfragments/66412.significant.rst
new file mode 100644
index 00000000000..22b06e7eeb0
--- /dev/null
+++ b/airflow-core/newsfragments/66412.significant.rst
@@ -0,0 +1,26 @@
+Fix triggerer race condition and deadlock that caused deferred tasks to stall
indefinitely
+
+Triggers that call synchronous SDK methods (e.g. ``get_task_states`` used by
+``safe_to_cancel`` in several Google provider operators) could crash the
triggerer's
+internal subprocess. The triggerer would then continue to heartbeat normally —
+appearing healthy to the scheduler — while silently processing zero triggers,
causing
+every deferred task to time out. This was first reported in
:github-issue:`64620`; a
+partial fix shipped in Airflow 3.2.1 (:github-pr:`64882`) but introduced a new
deadlock
+with the same visible symptom under load.
+
+Both issues are fixed by replacing the lock-based serialisation with response
+multiplexing: each request now carries a unique ID and the response is routed
back to
+the correct caller, so concurrent requests from trigger threads no longer
contend or
+deadlock regardless of how many triggers are running or what SDK methods they
call.
+
+**New: triggerer subprocess watchdog**
+
+Even with the race fixed, a trigger that blocks the event loop (e.g. by calling
+``time.sleep()`` or performing blocking I/O directly in ``async def run()``)
would
+previously leave the triggerer appearing healthy indefinitely.
+
+A new ``[triggerer] runner_health_check_threshold`` config option (default: 30
seconds)
+adds a watchdog: if the triggerer subprocess goes silent for longer than the
threshold,
+the parent process stops updating the heartbeat so the scheduler can detect
the hang and
+reassign triggers rather than waiting for them to individually time out. Set
the option
+to ``0`` to disable the watchdog.
diff --git a/airflow-core/src/airflow/config_templates/config.yml
b/airflow-core/src/airflow/config_templates/config.yml
index 445f89150db..46101ef2385 100644
--- a/airflow-core/src/airflow/config_templates/config.yml
+++ b/airflow-core/src/airflow/config_templates/config.yml
@@ -2650,6 +2650,17 @@ triggerer:
type: float
example: ~
default: "30"
+ runner_health_check_threshold:
+ description: |
+ If the TriggerRunner subprocess's async event loop sends no
communication to the parent
+ process for more than this many seconds, the parent stops updating the
triggerer's
+ heartbeat in the database. The triggerer then appears unhealthy to the
scheduler, which
+ will reassign its triggers. This detects a deadlocked or hung event
loop that the normal
+ process-alive check cannot catch. Set to 0 to disable the watchdog.
+ version_added: 3.2.2
+ type: float
+ example: ~
+ default: "30"
max_trigger_to_select_per_loop:
description: |
Maximum number of triggers to select per loop. Set this notably lower
than ``[triggerer] capacity``
diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
index a25632acac7..169455ac06f 100644
--- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
+++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
@@ -19,10 +19,12 @@ from __future__ import annotations
import asyncio
import functools
import logging
+import math
import os
import selectors
import signal
import sys
+import threading
import time
from collections import deque
from collections.abc import Generator, Iterable
@@ -34,6 +36,7 @@ from typing import TYPE_CHECKING, Annotated, Any, BinaryIO,
ClassVar, Literal, T
import anyio
import attrs
+import greenback
import structlog
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode
@@ -389,10 +392,19 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
queues: set[str] | None = None
health_check_threshold = conf.getint("triggerer",
"triggerer_health_check_threshold")
+ runner_health_check_threshold = conf.getfloat("triggerer",
"runner_health_check_threshold")
runner: TriggerRunner | None = None
stop: bool = False
+ # Timestamp of the last message received from the TriggerRunner
subprocess. Updated on
+ # every message; if it goes silent for longer than
runner_health_check_threshold the
+ # subprocess's async event loop has likely deadlocked. Initialised to
+inf so the watchdog
+ # stays silent until the first message arrives — avoids false positives on
slow/cold-start
+ # hosts where startup can exceed the threshold.
+ _last_runner_comms: float = attrs.field(init=False, default=math.inf)
+ _runner_comms_silence_logged: bool = attrs.field(init=False, default=False)
+
decoder: ClassVar[TypeAdapter[ToTriggerSupervisor]] =
TypeAdapter(ToTriggerSupervisor)
# Maps trigger IDs that we think are running in the sub process
@@ -451,6 +463,7 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
resp: BaseModel | None = None
dump_opts = {}
+ self._last_runner_comms = time.monotonic()
if isinstance(msg, messages.TriggerStateChanges):
if msg.events:
@@ -599,6 +612,19 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
self.emit_metrics()
def heartbeat(self):
+ elapsed = time.monotonic() - self._last_runner_comms
+ if self.runner_health_check_threshold > 0 and elapsed >
self.runner_health_check_threshold:
+ if not self._runner_comms_silence_logged:
+ log.error(
+ "TriggerRunner subprocess event loop appears deadlocked:
no communication received "
+ "for %.1fs (threshold: %ds). Skipping heartbeat so the
triggerer appears unhealthy "
+ "to the scheduler and its triggers are reassigned.",
+ elapsed,
+ self.runner_health_check_threshold,
+ )
+ self._runner_comms_silence_logged = True
+ return
+ self._runner_comms_silence_logged = False
perform_heartbeat(self.job,
heartbeat_callback=self.heartbeat_callback, only_if_necessary=True)
def heartbeat_callback(self, session: Session | None = None) -> None:
@@ -818,46 +844,80 @@ class TriggerCommsDecoder(CommsDecoder[ToTriggerRunner,
ToTriggerSupervisor]):
factory=lambda: TypeAdapter(ToTriggerRunner), repr=False
)
- def _read_frame(self):
- from asgiref.sync import async_to_sync
-
- with self._thread_lock:
- return async_to_sync(self._aread_frame)()
-
- 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)
+ _pending: dict[int, asyncio.Future] = attrs.field(factory=dict, repr=False)
+ _loop: asyncio.AbstractEventLoop | None = attrs.field(default=None,
repr=False)
+ _loop_thread_id: int | None = attrs.field(default=None, repr=False)
+ _reader_task: asyncio.Task | None = attrs.field(default=None, repr=False)
async def _aread_frame(self):
try:
len_bytes = await self._async_reader.readexactly(4)
except ConnectionResetError:
asyncio.current_task().cancel("Supervisor closed")
+ raise
length = int.from_bytes(len_bytes, byteorder="big")
if length >= 2**32:
raise OverflowError(f"Refusing to receive messages larger than
4GiB {length=}")
-
buffer = await self._async_reader.readexactly(length)
return self.resp_decoder.decode(buffer)
- async def _aget_response(self, expect_id: int) -> ToTriggerRunner | None:
- frame = await self._aread_frame()
- if frame.id != expect_id:
- # Given the lock we take out in `asend`, this _shouldn't_ be
possible, but I'd rather fail with
- # this explicit error return the wrong type of message back to a
Trigger
- raise RuntimeError(f"Response read out of order! Got {frame.id=},
{expect_id=}")
- return self._from_frame(frame)
+ async def _reader_loop(self) -> None:
+ try:
+ while True:
+ frame = await self._aread_frame()
+ future = self._pending.pop(frame.id, None)
+ if future is not None and not future.done():
+ future.set_result(frame)
+ else:
+ self.log.warning("Got response for unknown request frame",
frame_id=frame.id)
+ finally:
+ for fut in self._pending.values():
+ if not fut.done():
+ fut.cancel("Reader loop exited")
+ self._pending.clear()
- async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
- frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
- bytes = frame.as_bytes()
+ async def start_reader(self) -> None:
+ self._loop = asyncio.get_running_loop()
+ self._loop_thread_id = threading.get_ident()
+ self._reader_task = asyncio.create_task(self._reader_loop(),
name="trigger-comms-reader")
- async with self._async_lock:
- self._async_writer.write(bytes)
+ def send(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
+ if self._loop is None:
+ raise RuntimeError("start_reader() must be called before send()")
+ if threading.get_ident() == self._loop_thread_id:
+ # Called from the event loop thread itself (e.g. a trigger calling
a sync SDK method
+ # directly from async def run()).
run_coroutine_threadsafe(...).result() would deadlock
+ # here because .result() blocks the thread the event loop runs on.
+ # greenback.await_() teleports the coroutine back into the running
loop instead.
+ if not greenback.has_portal():
+ raise RuntimeError(
+ "Sync SDK methods (e.g. get_connection(), get_variable())
cannot be called "
+ "from a trigger's async def run() when
AIRFLOW_DISABLE_GREENBACK_PORTAL is "
+ "set. Either remove that environment variable, or use the
async equivalent "
+ "(e.g. aget_connection(), aget_variable())."
+ )
+ return greenback.await_(self.asend(msg))
+ return asyncio.run_coroutine_threadsafe(self.asend(msg),
self._loop).result()
- return await self._aget_response(frame.id)
+ async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
+ if self._loop is None:
+ raise RuntimeError("start_reader() must be called before asend()")
+ current_loop = asyncio.get_running_loop()
+ if self._loop is not None and current_loop is not self._loop:
+ # Called from a foreign event loop (e.g. via async_to_sync).
Bridge to the main loop
+ # so _reader_loop can resolve the future, then await via
wrap_future which is
+ # non-blocking for the foreign loop.
+ cf = asyncio.run_coroutine_threadsafe(self.asend(msg), self._loop)
+ return await asyncio.wrap_future(cf)
+ frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
+ future: asyncio.Future = current_loop.create_future()
+ self._pending[frame.id] = future
+ try:
+ self._async_writer.write(frame.as_bytes())
+ return self._from_frame(await future)
+ except BaseException:
+ self._pending.pop(frame.id, None)
+ raise
class TriggerRunner:
@@ -937,6 +997,10 @@ class TriggerRunner:
if watchdog.done():
watchdog.result()
+ if self.comms_decoder._reader_task.done():
+ self.comms_decoder._reader_task.result()
+ raise RuntimeError("Supervisor connection lost")
+
# Run core logic
finished_ids = await self.cleanup_finished_triggers()
@@ -960,6 +1024,11 @@ class TriggerRunner:
await log.aexception("Trigger runner failed")
self.stop = True
raise
+ finally:
+ if (reader_task := self.comms_decoder._reader_task) is not None:
+ reader_task.cancel()
+ with suppress(asyncio.CancelledError):
+ await reader_task
# Wait for supporting tasks to complete
await watchdog
@@ -982,11 +1051,13 @@ class TriggerRunner:
task_runner.SUPERVISOR_COMMS = self.comms_decoder
- msg = await self.comms_decoder._aget_response(expect_id=0)
-
+ frame = await self.comms_decoder._aread_frame()
+ msg = self.comms_decoder._from_frame(frame)
if not isinstance(msg, messages.StartTriggerer):
raise RuntimeError(f"Required first message to be a
messages.StartTriggerer, it was {msg}")
+ await self.comms_decoder.start_reader()
+
async def create_triggers(self):
"""Drain the to_create queue and create all new triggers that have
been requested in the DB."""
while self.to_create:
@@ -1205,8 +1276,6 @@ class TriggerRunner:
async def run_trigger(self, trigger_id: int, trigger: BaseTrigger,
timeout_after: datetime | None = None):
"""Run a trigger (they are async generators) and push their events
into our outbound event deque."""
if not os.environ.get("AIRFLOW_DISABLE_GREENBACK_PORTAL", "").lower()
== "true":
- import greenback
-
await greenback.ensure_portal()
ti = trigger.task_instance
diff --git a/airflow-core/tests/unit/jobs/test_triggerer_job.py
b/airflow-core/tests/unit/jobs/test_triggerer_job.py
index bf5d6de6ec8..550e9916e56 100644
--- a/airflow-core/tests/unit/jobs/test_triggerer_job.py
+++ b/airflow-core/tests/unit/jobs/test_triggerer_job.py
@@ -22,19 +22,24 @@ import contextlib
import datetime
import itertools
import os
+import random
import selectors
+import threading
import time
import typing
import uuid
from collections.abc import AsyncIterator
-from socket import socket
+from socket import socket, socketpair
from typing import TYPE_CHECKING, Any
from unittest import mock
from unittest.mock import ANY, AsyncMock, MagicMock, patch
+import greenback
+import msgspec
import pendulum
import pytest
-from asgiref.sync import sync_to_async
+import pytest_asyncio
+from asgiref.sync import async_to_sync, sync_to_async
from opentelemetry import trace as otel_trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
@@ -67,7 +72,7 @@ from airflow.providers.standard.operators.python import
PythonOperator
from airflow.providers.standard.triggers.file import FileDeleteTrigger
from airflow.providers.standard.triggers.temporal import DateTimeTrigger,
TimeDeltaTrigger
from airflow.sdk import DAG, BaseHook, BaseOperator
-from airflow.sdk.execution_time.comms import ToSupervisor, ToTask
+from airflow.sdk.execution_time.comms import ToSupervisor, ToTask,
_RequestFrame, _ResponseFrame
from airflow.serialization.serialized_objects import LazyDeserializedDAG
from airflow.triggers.base import BaseTrigger, TriggerEvent
from airflow.triggers.testing import FailureTrigger, SuccessTrigger
@@ -229,6 +234,54 @@ def supervisor_builder(mocker, session):
return builder
+def test_heartbeat_watchdog(supervisor_builder, mocker):
+ """heartbeat() fires when the subprocess is active, skips when silent, and
only arms the
+ silence flag once (so the error is logged once, not on every subsequent
call)."""
+ supervisor = supervisor_builder()
+ perform_heartbeat_mock =
mocker.patch("airflow.jobs.triggerer_job_runner.perform_heartbeat")
+
+ # Within threshold — heartbeat fires
+ supervisor._last_runner_comms = time.monotonic() - 5.0
+ supervisor.heartbeat()
+ perform_heartbeat_mock.assert_called_once()
+
+ # Just inside threshold (29.9s < 30s default) — still fires
+ supervisor._last_runner_comms = time.monotonic() - 29.9
+ supervisor.heartbeat()
+ assert perform_heartbeat_mock.call_count == 2
+
+ # Beyond threshold — heartbeat skips and silence flag is set
+ supervisor._last_runner_comms = time.monotonic() - 9999.0
+ supervisor.heartbeat()
+ assert perform_heartbeat_mock.call_count == 2
+ assert supervisor._runner_comms_silence_logged is True
+
+ # Subsequent silent heartbeats don't re-arm the flag (error logged only
once)
+ supervisor.heartbeat()
+ supervisor.heartbeat()
+ assert perform_heartbeat_mock.call_count == 2
+ assert supervisor._runner_comms_silence_logged is True
+
+ # Once the subprocess speaks again the flag resets and heartbeat resumes
+ supervisor._last_runner_comms = time.monotonic()
+ supervisor.heartbeat()
+ assert perform_heartbeat_mock.call_count == 3
+ assert supervisor._runner_comms_silence_logged is False
+
+
+def
test_heartbeat_watchdog_disabled_when_threshold_is_zero(supervisor_builder,
mocker):
+ """Setting runner_health_check_threshold=0 disables the watchdog;
heartbeat always fires."""
+ supervisor = supervisor_builder()
+ mocker.patch.object(type(supervisor), "runner_health_check_threshold",
new=0)
+ perform_heartbeat_mock =
mocker.patch("airflow.jobs.triggerer_job_runner.perform_heartbeat")
+
+ supervisor._last_runner_comms = time.monotonic() - 9999.0
+
+ supervisor.heartbeat()
+
+ perform_heartbeat_mock.assert_called_once()
+
+
def test_trigger_lifecycle(spy_agency: SpyAgency, session, testing_dag_bundle):
"""
Checks that the triggerer will correctly see a new Trigger in the database
@@ -1455,3 +1508,175 @@ class TestMakeTriggerSpan:
assert attrs["airflow.trigger.name"] == "OnlyTrigger"
assert "airflow.dag_id" not in attrs
assert "airflow.task_id" not in attrs
+
+
+def _read_frame_sync(sock) -> _RequestFrame | None:
+ """Read a length-prefixed msgpack frame from a blocking socket."""
+ lb = b""
+ while len(lb) < 4:
+ chunk = sock.recv(4 - len(lb))
+ if not chunk:
+ return None
+ lb += chunk
+ n = int.from_bytes(lb, "big")
+ data = b""
+ while len(data) < n:
+ chunk = sock.recv(n - len(data))
+ if not chunk:
+ return None
+ data += chunk
+ return msgspec.msgpack.decode(data, type=_RequestFrame)
+
+
+@pytest_asyncio.fixture
+async def decoder_pair():
+ """Yield (decoder, server_sock). Caller owns closing."""
+ server_sock, client_sock = socketpair()
+ reader, writer = await asyncio.open_connection(sock=client_sock)
+ decoder = TriggerCommsDecoder(async_writer=writer, async_reader=reader,
socket=client_sock)
+ await decoder.start_reader()
+ yield decoder, server_sock
+ if decoder._reader_task:
+ if not decoder._reader_task.done():
+ decoder._reader_task.cancel()
+ with contextlib.suppress(asyncio.CancelledError, Exception):
+ await decoder._reader_task
+ writer.close()
+ server_sock.close()
+
+
[email protected]
[email protected]_timeout(15)
+async def test_all_send_paths_concurrent(decoder_pair):
+ """
+ All four send() paths running concurrently with responses returned out of
order:
+
+ 1. asend() directly from async code — pure-async path
+ 2. send() via asyncio.to_thread() — mirrors
apache/airflow#63913:
+
sync_to_async(hook_class)() → get_connection()
+ →
SUPERVISOR_COMMS.send() from a thread pool thread
+ 3. send() from the event-loop thread — mirrors
apache/airflow#63760:
+ via greenback async_to_sync raised
RuntimeError in same thread
+ 4. async_to_sync(asend)() from a thread — trigger code that wraps
an async fn which
+ internally calls asend;
bridges via wrap_future
+
+ The concurrent mix with shuffled responses also covers
apache/airflow#65286: the
+ _thread_lock + async_to_sync approach stalled the triggerer under this
exact load pattern.
+ """
+ decoder, server_sock = decoder_pair
+ N = 5
+ N_TOTAL = N * 4
+
+ def supervisor():
+ frames = []
+ for _ in range(N_TOTAL):
+ f = _read_frame_sync(server_sock)
+ if f is None:
+ break
+ frames.append(f)
+ random.shuffle(frames)
+ for f in frames:
+ server_sock.sendall(
+ _ResponseFrame(
+ id=f.id,
+ body={"type": "TriggerStateSync", "to_create": [],
"to_cancel": []},
+ ).as_bytes()
+ )
+
+ sup = threading.Thread(target=supervisor, daemon=True)
+ sup.start()
+
+ async def async_send(idx):
+ return await decoder.asend(messages.TriggerStateChanges(events=None,
finished=[idx], failures=None))
+
+ async def from_thread_send(idx):
+ # In production this path is taken by asgiref's own thread pool
(sync_to_async),
+ # which is invisible to asyncio's default executor. We avoid
asyncio.to_thread()
+ # here because on Python < 3.12 loop.shutdown_default_executor() has
no timeout
+ # and hangs if any executor threads are still alive at loop teardown.
+ # TODO: simplify with asyncio.to_thread() when Python 3.12 is the
minimum.
+ loop = asyncio.get_running_loop()
+ fut: asyncio.Future[messages.TriggerStateSync] = loop.create_future()
+
+ def sync_send():
+ try:
+ result = decoder.send(
+ messages.TriggerStateChanges(events=None, finished=[N +
idx], failures=None)
+ )
+ loop.call_soon_threadsafe(fut.set_result, result)
+ except Exception as exc:
+ loop.call_soon_threadsafe(fut.set_exception, exc)
+
+ threading.Thread(target=sync_send, daemon=True).start()
+ return await fut
+
+ async def greenback_send(idx):
+ await greenback.ensure_portal()
+ return decoder.send(messages.TriggerStateChanges(events=None,
finished=[2 * N + idx], failures=None))
+
+ async def async_to_sync_send(idx):
+ # Same executor-avoidance reason as from_thread_send above.
+ # TODO: simplify with asyncio.to_thread() when Python 3.12 is the
minimum.
+ loop = asyncio.get_running_loop()
+ fut: asyncio.Future[messages.TriggerStateSync] = loop.create_future()
+
+ def thread_fn():
+ try:
+ result = async_to_sync(decoder.asend)(
+ messages.TriggerStateChanges(events=None, finished=[3 * N
+ idx], failures=None)
+ )
+ loop.call_soon_threadsafe(fut.set_result, result)
+ except Exception as exc:
+ loop.call_soon_threadsafe(fut.set_exception, exc)
+
+ threading.Thread(target=thread_fn, daemon=True).start()
+ return await fut
+
+ results = await asyncio.gather(
+ *[asyncio.create_task(async_send(i)) for i in range(N)],
+ *[asyncio.create_task(from_thread_send(i)) for i in range(N)],
+ *[asyncio.create_task(greenback_send(i)) for i in range(N)],
+ *[asyncio.create_task(async_to_sync_send(i)) for i in range(N)],
+ return_exceptions=True,
+ )
+
+ sup.join(timeout=5)
+
+ errors = [r for r in results if isinstance(r, Exception)]
+ assert not errors, f"errors: {errors}"
+ assert len(results) == N_TOTAL
+ assert all(isinstance(r, messages.TriggerStateSync) for r in results)
+
+
[email protected]
+async def test_connection_close_cancels_pending(decoder_pair):
+ """When the connection closes while asend() is awaiting, the future is
cancelled."""
+ decoder, server_sock = decoder_pair
+
+ task = asyncio.create_task(
+ decoder.asend(messages.TriggerStateChanges(events=None, finished=[1],
failures=None))
+ )
+ await asyncio.sleep(0)
+
+ server_sock.close()
+
+ with pytest.raises((asyncio.CancelledError, Exception)):
+ await asyncio.wait_for(task, timeout=5)
+
+
[email protected]
+async def test_unknown_frame_id_doesnt_crash_reader(decoder_pair):
+ """An orphan response frame (no matching pending future) is silently
dropped; reader stays alive."""
+ decoder, server_sock = decoder_pair
+
+ server_sock.sendall(
+ _ResponseFrame(
+ id=99999,
+ body={"type": "TriggerStateSync", "to_create": [], "to_cancel":
[]},
+ ).as_bytes()
+ )
+
+ await asyncio.sleep(0.05)
+
+ assert decoder._reader_task is not None
+ assert not decoder._reader_task.done(), "reader loop crashed unexpectedly"
diff --git a/task-sdk/src/airflow/sdk/execution_time/secrets/execution_api.py
b/task-sdk/src/airflow/sdk/execution_time/secrets/execution_api.py
index 6bda6440e07..b8b240b4988 100644
--- a/task-sdk/src/airflow/sdk/execution_time/secrets/execution_api.py
+++ b/task-sdk/src/airflow/sdk/execution_time/secrets/execution_api.py
@@ -65,26 +65,6 @@ class ExecutionAPISecretsBackend(BaseSecretsBackend):
# Convert ExecutionAPI response to SDK Connection
return _process_connection_result_conn(msg)
- except RuntimeError as e:
- # TriggerCommsDecoder.send() uses async_to_sync internally, which
raises RuntimeError
- # when called within an async event loop. In greenback portal
contexts (triggerer),
- # we catch this and use greenback to call the async version
instead.
- if str(e).startswith("You cannot use AsyncToSync in the same
thread as an async event loop"):
- import asyncio
-
- import greenback
-
- task = asyncio.current_task()
- if greenback.has_portal(task):
- import warnings
-
- warnings.warn(
- "You should not use sync calls here -- use `await
aget_connection` instead",
- stacklevel=2,
- )
- return greenback.await_(self.aget_connection(conn_id))
- # Fall through to the general exception handler for other
RuntimeErrors
- return None
except Exception:
# If SUPERVISOR_COMMS fails for any reason, return None
# to allow fallback to other backends
@@ -113,26 +93,6 @@ class ExecutionAPISecretsBackend(BaseSecretsBackend):
if isinstance(msg, VariableResult):
return msg.value # Already a string | None
return None
- except RuntimeError as e:
- # TriggerCommsDecoder.send() uses async_to_sync internally, which
raises RuntimeError
- # when called within an async event loop. In greenback portal
contexts (triggerer),
- # we catch this and use greenback to call the async version
instead.
- if str(e).startswith("You cannot use AsyncToSync in the same
thread as an async event loop"):
- import asyncio
-
- import greenback
-
- task = asyncio.current_task()
- if greenback.has_portal(task):
- import warnings
-
- warnings.warn(
- "You should not use sync calls here -- use `await
aget_variable` instead",
- stacklevel=2,
- )
- return greenback.await_(self.aget_variable(key))
- # Fall through to the general exception handler for other
RuntimeErrors
- return None
except Exception:
# If SUPERVISOR_COMMS fails for any reason, return None
# to allow fallback to other backends
diff --git a/task-sdk/tests/task_sdk/execution_time/test_secrets.py
b/task-sdk/tests/task_sdk/execution_time/test_secrets.py
index ccdabb3dd58..bda87a6a64a 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_secrets.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_secrets.py
@@ -19,7 +19,6 @@ from __future__ import annotations
import pytest
-from airflow.sdk.definitions.connection import Connection
from airflow.sdk.execution_time.secrets.execution_api import
ExecutionAPISecretsBackend
@@ -121,106 +120,6 @@ class TestExecutionAPISecretsBackend:
with pytest.raises(NotImplementedError, match="Use get_connection
instead"):
backend.get_conn_value("test_conn")
- def test_runtime_error_triggers_greenback_fallback(self, mocker,
mock_supervisor_comms):
- """
- Test that RuntimeError from async_to_sync triggers greenback fallback.
-
- This test verifies the fix for issue #57145: when
SUPERVISOR_COMMS.send()
- raises the specific RuntimeError about async_to_sync in an event loop,
- the backend catches it and uses greenback to call aget_connection().
- """
-
- # Expected connection to be returned
- expected_conn = Connection(
- conn_id="databricks_default",
- conn_type="databricks",
- host="example.databricks.com",
- )
-
- # Simulate the RuntimeError that triggers greenback fallback
- mock_supervisor_comms.send.side_effect = RuntimeError(
- "You cannot use AsyncToSync in the same thread as an async event
loop"
- )
-
- # Mock the greenback and asyncio modules that are imported inside the
exception handler
- mocker.patch("greenback.has_portal", return_value=True)
- mocker.patch("asyncio.current_task")
-
- # Mock greenback.await_ to actually await the coroutine it receives.
- # This prevents Python 3.13 RuntimeWarning about unawaited coroutines.
- import asyncio
-
- def greenback_await_side_effect(coro):
- loop = asyncio.new_event_loop()
- try:
- return loop.run_until_complete(coro)
- finally:
- loop.close()
-
- mock_greenback_await = mocker.patch("greenback.await_",
side_effect=greenback_await_side_effect)
-
- # Mock aget_connection to return the expected connection directly.
- # We need to mock this because the real aget_connection would try to
- # use SUPERVISOR_COMMS.asend which is not set up for this test.
- async def mock_aget_connection(self, conn_id):
- return expected_conn
-
- mocker.patch.object(ExecutionAPISecretsBackend, "aget_connection",
mock_aget_connection)
-
- backend = ExecutionAPISecretsBackend()
- conn = backend.get_connection("databricks_default")
-
- # Verify we got the expected connection
- assert conn is not None
- assert conn.conn_id == "databricks_default"
- # Verify the greenback fallback was called
- mock_greenback_await.assert_called_once()
- # Verify send was attempted first (and raised RuntimeError)
- mock_supervisor_comms.send.assert_called_once()
-
- def test_get_variable_runtime_error_triggers_greenback_fallback(self,
mocker, mock_supervisor_comms):
- """
- Test that RuntimeError from async_to_sync triggers greenback fallback
for variables.
-
- Same as the connection test but for get_variable — verifies the fix
for #61676:
- triggers calling Variable.get() fail because SUPERVISOR_COMMS.send()
raises
- RuntimeError in the async event loop, but the greenback fallback was
missing.
- """
- expected_value = "10"
-
- # Simulate the RuntimeError that triggers greenback fallback
- mock_supervisor_comms.send.side_effect = RuntimeError(
- "You cannot use AsyncToSync in the same thread as an async event
loop"
- )
-
- # Mock the greenback and asyncio modules
- mocker.patch("greenback.has_portal", return_value=True)
- mocker.patch("asyncio.current_task")
-
- import asyncio
-
- def greenback_await_side_effect(coro):
- loop = asyncio.new_event_loop()
- try:
- return loop.run_until_complete(coro)
- finally:
- loop.close()
-
- mock_greenback_await = mocker.patch("greenback.await_",
side_effect=greenback_await_side_effect)
-
- # Mock aget_variable to return the expected value
- async def mock_aget_variable(self, key):
- return expected_value
-
- mocker.patch.object(ExecutionAPISecretsBackend, "aget_variable",
mock_aget_variable)
-
- backend = ExecutionAPISecretsBackend()
- result = backend.get_variable("retries")
-
- assert result == expected_value
- mock_greenback_await.assert_called_once()
- mock_supervisor_comms.send.assert_called_once()
-
class TestContextDetection:
"""Test context detection in ensure_secrets_backend_loaded."""
