This is an automated email from the ASF dual-hosted git repository.
ephraimbuddy pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new ddaf81ccff6 Extract run_once/should_stop/run_context from
TriggerRunnerSupervisor.run() (#65675)
ddaf81ccff6 is described below
commit ddaf81ccff6f1223c6b93d555745560c6b27b9eb
Author: Ephraim Anierobi <[email protected]>
AuthorDate: Mon Apr 27 12:00:36 2026 +0100
Extract run_once/should_stop/run_context from TriggerRunnerSupervisor.run()
(#65675)
In preparation for AIP-92, split the run loop into overridable
seams so alternative Triggerer backends can swap per-iteration work,
the termination condition, and loop setup/teardown without replacing
run(). No behavior change.
---
.../src/airflow/jobs/triggerer_job_runner.py | 48 ++++++++++++-------
airflow-core/tests/unit/jobs/test_triggerer_job.py | 56 ++++++++++++++++++++++
2 files changed, 87 insertions(+), 17 deletions(-)
diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
index 45a6c7ddc0d..73c0ed906e8 100644
--- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
+++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
@@ -25,8 +25,8 @@ import signal
import sys
import time
from collections import deque
-from collections.abc import Callable, Generator, Iterable
-from contextlib import suppress
+from collections.abc import Callable, Generator, Iterable, Iterator
+from contextlib import contextmanager, suppress
from datetime import datetime
from socket import socket
from traceback import format_exception
@@ -580,21 +580,35 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
def run(self) -> None:
"""Run synchronously and handle all database reads/writes."""
- while not self.stop:
- if not self.is_alive():
- log.error("Trigger runner process has died! Exiting.")
- break
- self.load_triggers()
-
- # Wait for up to 1 second for activity
- self._service_subprocess(1)
-
- self.handle_events()
- self.handle_failed_triggers()
- self.clean_unused()
- self.heartbeat()
-
- self.emit_metrics()
+ with self.run_context():
+ while not self.should_stop():
+ if not self.is_alive():
+ log.error("Trigger runner process has died! Exiting.")
+ break
+ self.run_once()
+
+ @contextmanager
+ def run_context(self) -> Iterator[None]:
+ """Wrap the run loop. Subclasses can override to install
setup/teardown."""
+ yield
+
+ def should_stop(self) -> bool:
+ """Return True when the run loop should exit."""
+ return self.stop
+
+ def run_once(self) -> None:
+ """Perform a single iteration of the run loop."""
+ self.load_triggers()
+
+ # Wait for up to 1 second for activity
+ self._service_subprocess(1)
+
+ self.handle_events()
+ self.handle_failed_triggers()
+ self.clean_unused()
+ self.heartbeat()
+
+ self.emit_metrics()
def heartbeat(self):
perform_heartbeat(self.job,
heartbeat_callback=self.heartbeat_callback, only_if_necessary=True)
diff --git a/airflow-core/tests/unit/jobs/test_triggerer_job.py
b/airflow-core/tests/unit/jobs/test_triggerer_job.py
index 08c960fd33f..4b047ab15e1 100644
--- a/airflow-core/tests/unit/jobs/test_triggerer_job.py
+++ b/airflow-core/tests/unit/jobs/test_triggerer_job.py
@@ -229,6 +229,62 @@ def supervisor_builder(mocker, session):
return builder
+def test_run_invokes_seams_in_order(supervisor_builder, mocker):
+ """run() enters run_context, drives run_once while not should_stop, then
exits run_context."""
+ from contextlib import contextmanager
+
+ supervisor = supervisor_builder()
+ events: list[str] = []
+
+ @contextmanager
+ def fake_run_context(self):
+ events.append("enter")
+ try:
+ yield
+ finally:
+ events.append("exit")
+
+ counter = {"n": 0}
+
+ def fake_run_once(self):
+ counter["n"] += 1
+ events.append(f"tick-{counter['n']}")
+
+ mocker.patch.object(TriggerRunnerSupervisor, "run_context",
fake_run_context)
+ mocker.patch.object(TriggerRunnerSupervisor, "run_once", fake_run_once)
+ mocker.patch.object(TriggerRunnerSupervisor, "should_stop",
side_effect=lambda: counter["n"] >= 3)
+ mocker.patch.object(TriggerRunnerSupervisor, "is_alive", return_value=True)
+
+ supervisor.run()
+
+ assert events == ["enter", "tick-1", "tick-2", "tick-3", "exit"]
+
+
+def test_run_context_exits_when_subprocess_dies(supervisor_builder, mocker):
+ """Breaking out of the loop on a dead subprocess still unwinds
run_context."""
+ from contextlib import contextmanager
+
+ supervisor = supervisor_builder()
+ events: list[str] = []
+
+ @contextmanager
+ def fake_run_context(self):
+ events.append("enter")
+ try:
+ yield
+ finally:
+ events.append("exit")
+
+ mocker.patch.object(TriggerRunnerSupervisor, "run_context",
fake_run_context)
+ mocker.patch.object(TriggerRunnerSupervisor, "run_once",
side_effect=lambda: events.append("tick"))
+ mocker.patch.object(TriggerRunnerSupervisor, "should_stop",
return_value=False)
+ mocker.patch.object(TriggerRunnerSupervisor, "is_alive",
side_effect=[True, False])
+
+ supervisor.run()
+
+ assert events == ["enter", "tick", "exit"]
+
+
def test_trigger_lifecycle(spy_agency: SpyAgency, session, testing_dag_bundle):
"""
Checks that the triggerer will correctly see a new Trigger in the database
