Avoid scheduler crash when periodic maintenance actions fail

[hanxdatadog]

What does this PR do?

• Adds an opt-in non_fatal: bool = False parameter to EventScheduler.call_regular_interval. When True, an exception raised by the periodic action is logged (self.log.warning(..., exc_info=True)) and swallowed instead of propagating; the next cycle is still scheduled either way. Default is False, preserving current behavior for existing callers.
• Enables non_fatal=True for the two SchedulerJobRunner periodic maintenance callbacks registered via call_regular_interval that had no exception handling at all: _remove_unreferenced_triggers and _reap_stale_connection_tests.
• Logs Failed to run periodic action <name> due to <exception>; will retry on the next cycle on the caught-exception path — matches the existing Failed to <thing> due to %s house style already used by _update_dag_run_state_for_paused_dags, and states why a single occurrence is safe to ignore. Logged at WARNING (with exc_info=True to still capture the full traceback) rather than ERROR, so a persistent (rather than transient) failure repeating every cycle doesn't generate steady-state ERROR-level noise.

Motivation:
EventScheduler.call_regular_interval's inner repeat() closure calls the registered action with no exception handling, so a single transient failure (e.g. a DB statement_timeout during _remove_unreferenced_triggers's DELETE) propagates all the way up and crashes the whole SchedulerJob. In practice this means: the scheduler process dies with a critical-level log, and every task instance it was tracking gets swept into the orphaned-task-adoption path on the next scheduler startup — a heavyweight, disruptive recovery flow triggered by what should have been a harmless, skippable cycle of a periodic cleanup job. This is inconsistent with the existing pattern elsewhere in SchedulerJobRunner — _update_dag_run_state_for_paused_dags already wraps its own body in except Exception as e: # should not fail the scheduler, and the same crash-avoidance philosophy has precedent in merged PRs like #62893 (isolating per-DagRun failures so one bad DagRun can't crash the scheduler). EventScheduler is documented as a "general purpose" utility, so rather than making it always swallow exceptions (which would silently change behavior for any future caller that wants failures to propagate), the fix is an explicit opt-in.

Testing:

• Unit tests: 2 new tests in test_event_scheduler.py — one confirms the default (no flag) still propagates the action's exception, one confirms non_fatal=True swallows it and still schedules the next cycle.

Notes:

• Named non_fatal rather than catch_exceptions to be outcome-focused (this failure isn't fatal to the scheduler) rather than mechanism-focused, matching existing "non-fatal" terminology already used elsewhere in the codebase (dag_processing/importers/base.py).
• Log level was changed from ERROR (self.log.exception) to WARNING (self.log.warning(..., exc_info=True)) in response to review feedback — addresses steady-state log noise for a persistent failure while keeping the traceback for debuggability.

---

Was generative AI tooling used to co-author this PR?

• Yes (Claude Code)

[Vamsi-klu]

Confirmed the root cause: repeat() in call_regular_interval calls action() with no exception handling, and both maintenance callbacks it guards -- _remove_unreferenced_triggers (a raw DELETE with synchronize_session=fetch) and _reap_stale_connection_tests (a row-locked SELECT plus UPDATE loop and flush) -- have no internal error handling. timers.run(blocking=False) is called inside the while loop of _run_scheduler_loop with no try/except, so a transient DB error (statement_timeout, deadlock, dropped connection) propagates out and takes down the SchedulerJob for what should be a skippable cleanup cycle. The opt-in flag is the right call since EventScheduler is general-purpose and default=False keeps every existing caller identical. except Exception (not bare except) correctly still lets KeyboardInterrupt/SystemExit through, and self.enter() is the last line of repeat() outside the try, so the next cycle is always rescheduled. Worth noting: both callbacks are @provide_session-decorated, so the session is already rolled back and closed by the time the exception reaches the except -- swallowing doesn't leave a half-open transaction. This matches the existing precedent in _update_dag_run_state_for_paused_dags. Regression coverage is meaningful: the non_fatal test fails without the guard (asserts no raise plus queue growth 1 to 2), and the default-path test pins that the unguarded behavior still propagates.

One non-blocking question: self.log.exception logs a full traceback at ERROR every cycle, so a persistent (rather than transient) failure will repeat the traceback each interval and the cleanup will silently never succeed. Given the coarse interval that's probably acceptable, but did you consider log.warning without the traceback, or logging once, to avoid steady-state log noise?

[hanxdatadog]

Confirmed the root cause: repeat() in call_regular_interval calls action() with no exception handling, and both maintenance callbacks it guards -- _remove_unreferenced_triggers (a raw DELETE with synchronize_session=fetch) and _reap_stale_connection_tests (a row-locked SELECT plus UPDATE loop and flush) -- have no internal error handling. timers.run(blocking=False) is called inside the while loop of _run_scheduler_loop with no try/except, so a transient DB error (statement_timeout, deadlock, dropped connection) propagates out and takes down the SchedulerJob for what should be a skippable cleanup cycle. The opt-in flag is the right call since EventScheduler is general-purpose and default=False keeps every existing caller identical. except Exception (not bare except) correctly still lets KeyboardInterrupt/SystemExit through, and self.enter() is the last line of repeat() outside the try, so the next cycle is always rescheduled. Worth noting: both callbacks are @provide_session-decorated, so the session is already rolled back and closed by the time the exception reaches the except -- swallowing doesn't leave a half-open transaction. This matches the existing precedent in _update_dag_run_state_for_paused_dags. Regression coverage is meaningful: the non_fatal test fails without the guard (asserts no raise plus queue growth 1 to 2), and the default-path test pins that the unguarded behavior still propagates.

One non-blocking question: self.log.exception logs a full traceback at ERROR every cycle, so a persistent (rather than transient) failure will repeat the traceback each interval and the cleanup will silently never succeed. Given the coarse interval that's probably acceptable, but did you consider log.warning without the traceback, or logging once, to avoid steady-state log noise?

Hi @Vamsi-klu Thank you for the comment. I just made the log at the warning level. For the change's motivation, exactly as you described, we hit some crashes in some supposably non-fatal cleaning up schedule actions. here is to provide some general option for these kind of non-fatal scheduled actions. With the flag, the goal is to not let scheduler crash (making some tasks enter orphan adoption/drop path unnecessarily) meanwhile giving the maintainers enough attentions to take actions to address them shortly (warning level logs with stacktrace)