[v3-0-test] Handle trigger calls to get_connection apache#55799

In 2.x sometimes get_connection (which goes to the database) might be called without wrapping in sync_to_async.

This did not fail, though it was not good behavior, since it can block the event loop.

In 3.0, since we now route db calls through an API, triggers that do this fail. The reason is, the code to hit the API wraps the get_connection call with async_to_sync, which is forbidden in the asyncio event loop.

Related: apache#55568

(cherry picked from commit f5b1eb4)

Author: dstandish

airflow-core/pyproject.toml

@@ -153,6 +153,7 @@ dependencies = [
 "async" = [
     "eventlet>=0.37.0",
     "gevent>=24.2.1",
+    "greenback>=1.2.1",
     "greenlet>=0.4.9",
 ]
 "graphviz" = [

airflow-core/src/airflow/jobs/triggerer_job_runner.py

@@ -737,7 +737,24 @@ def _read_frame(self):
     def send(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
         from asgiref.sync import async_to_sync
 
-        return async_to_sync(self.asend)(msg)
+        try:
+            return async_to_sync(self.asend)(msg)
+        except RuntimeError as e:
+            if str(e).startswith("You cannot use AsyncToSync in the same thread as an async event loop"):
+                log.warning(
+                    "`TriggerCommsDecoder.send` is being called from within a coroutine without "
+                    "having been wrapped with sync_to_async; attempting to invoke with greenback. "
+                    "You must find where this in being invoked and wrap with sync_to_async."
+                )
+                try:
+                    import greenback
+
+                    task = asyncio.current_task()
+                    if greenback.has_portal(task):
+                        return greenback.await_(self.asend(msg))
+                except ImportError:
+                    log.warning("greenback unavailable; raising")
+            raise
 
     async def _aread_frame(self):
         len_bytes = await self._async_reader.readexactly(4)
@@ -1064,6 +1081,15 @@ async def block_watchdog(self):
 
     async def run_trigger(self, trigger_id, trigger):
         """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":
+            try:
+                import greenback
+
+                await greenback.ensure_portal()
+
+            except ImportError:
+                log.warning("greenback unavailable; not creating portal.")
+
         bind_log_contextvars(trigger_id=trigger_id)
 
         name = self.triggers[trigger_id]["name"]