Avoid scheduler crash when periodic maintenance actions fail

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

@@ -1747,6 +1747,7 @@ def _run_scheduler_loop(self) -> None:
         timers.call_regular_interval(
             conf.getfloat("scheduler", "parsing_cleanup_interval"),
             self._remove_unreferenced_triggers,
+            non_fatal=True,
         )
 
         if any(x.is_local for x in self.executors):
@@ -1764,6 +1765,7 @@ def _run_scheduler_loop(self) -> None:
         timers.call_regular_interval(
             delay=conf.getfloat("connection_test", "reaper_interval", fallback=30.0),
             action=self._reap_stale_connection_tests,
+            non_fatal=True,
         )
 
         idle_count = 0

airflow-core/src/airflow/utils/event_scheduler.py

@@ -32,12 +32,31 @@ def call_regular_interval(
         action: Callable,
         arguments=(),
         kwargs=None,
+        non_fatal: bool = False,
     ):
-        """Call a function at (roughly) a given interval."""
+        """
+        Call a function at (roughly) a given interval.
+
+        :param non_fatal: If True, an exception raised by ``action`` is logged and
+            swallowed instead of propagating, so a single bad cycle can't kill
+            whatever is driving this scheduler. The next cycle is still scheduled either
+            way. Defaults to False (propagate), preserving prior behavior for callers
+            that rely on the exception surfacing.
+        """
 
         def repeat(*args, **kwargs):
             self.log.debug("Calling %s", action)
-            action(*args, **kwargs)
+            if non_fatal:
+                try:
+                    action(*args, **kwargs)
+                except Exception as e:
+                    self.log.exception(
+                        "Failed to run periodic action %s due to %s; will retry on the next cycle",
+                        getattr(action, "__name__", action),
+                        e,
+                    )
+            else:
+                action(*args, **kwargs)
             # This is not perfect. If we want a timer every 60s, but action
             # takes 10s to run, this will run it every 70s.
             # Good enough for now

airflow-core/tests/unit/utils/test_event_scheduler.py

@@ -19,6 +19,8 @@
 
 from unittest import mock
 
+import pytest
+
 from airflow.utils.event_scheduler import EventScheduler
 
 
@@ -38,3 +40,34 @@ def test_call_regular_interval(self):
         assert len(timers.queue) == 2
         somefunction.assert_called_once()
         assert timers.queue[0].time < timers.queue[1].time
+
+    def test_call_regular_interval_propagates_exception_by_default(self):
+        """Without opting in, an action's exception still propagates (unchanged default behavior)."""
+        failing_action = mock.MagicMock(side_effect=RuntimeError("boom"))
+
+        timers = EventScheduler()
+        timers.call_regular_interval(30, failing_action)
+        assert len(timers.queue) == 1
+
+        with pytest.raises(RuntimeError, match="boom"):
+            timers.queue[0].action()
+
+        failing_action.assert_called_once()
+        # The next cycle was never scheduled because the exception propagated.
+        assert len(timers.queue) == 1
+
+    def test_call_regular_interval_non_fatal_swallows_action_exception(self):
+        """With non_fatal=True, a raising action is swallowed and the next cycle is still scheduled."""
+        failing_action = mock.MagicMock(side_effect=RuntimeError("boom"))
+
+        timers = EventScheduler()
+        timers.call_regular_interval(30, failing_action, non_fatal=True)
+        assert len(timers.queue) == 1
+
+        # Should not raise, even though the action does.
+        timers.queue[0].action()
+
+        failing_action.assert_called_once()
+        # The next cycle was still scheduled despite the exception.
+        assert len(timers.queue) == 2
+        assert timers.queue[0].time < timers.queue[1].time