[seanghaeli]

A trigger that keeps failing leaves nothing in the database, so no API, UI or alert can tell, and the row doesn't say who the trigger runs for. triggers.failed can't tell a watcher from a task's trigger either.

• Each failure writes consecutive_failures, first_failed_at, last_failed_at and last_error to the trigger's row. last_error is the exception type and message only, masked like task logs and cut to 1,000 characters. An event sets last_event_at.
• An error message is stored with any characters the database can't hold (such as NUL on Postgres) escaped, so a bad message can't stop the triggerer.
• A trigger counts as healthy again, and its failure count resets, when it fires an event or runs for two minutes without failing (Trigger recovered is logged). So a watcher that ran fine for a day and then crashed once shows one failure, not every failure since it first crashed.
• Task and callback triggers record dag_id and owner_ref when created; the migration adds nullable columns and fills the owner in for existing rows, and the triggerer fills it in for rows older components create during a rolling upgrade.
• triggers.failed and triggers.succeeded get a trigger_kind tag (one of task, asset_watcher or callback), by what the trigger runs for. A new triggers.failing gauge counts failing triggers per triggerer and kind.

How to review: TriggerFailureTracker in jobs/triggerer_job_runner.py, then the row writes, Trigger.submit_failure and clear_recorded_failures in models/trigger.py.

Breeze on Postgres, a watcher with expired credentials next to healthy watchers, deferred tasks and a deadline callback:

# after this PR
=== A) [09:39:21] creds watcher failing, re-created as on main
--- row: (id, consecutive_failures, first_failed_at, last_failed_at, last_event_at, last_error)
(1, 36, '09:38:16', '09:39:27', None, 'ConnectionError: TO creds: the security token included in the request is expired')
  creds instance starts: 16 (bad=True 16); gaps s: [2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0] ...
  triggers.failing (latest per kind): asset_watcher=1 callback=0 task=0 
  triggers.failed lines by tag:  37 triggers.failed,trigger_kind=asset_watcher; 1 triggers.failed,trigger_kind=task;
=== deferred task trigger that raised: wait_fail ti state=failed, trigger instances started=1, last poll=TO-TASKWAIT wait_fail poll 20, task triggers left on rows: 0
  triggers.failing (latest per kind): asset_watcher=1 callback=0 task=0 
  triggers.failed lines by tag:  40 triggers.failed,trigger_kind=asset_watcher; 1 triggers.failed,trigger_kind=task;
=== B) [09:39:37] credentials fixed and the watcher fires: recovery by an event
--- row:
(1, None, None, '09:39:37', '09:39:39', 'ConnectionError: TO creds: the security token included in the request is expired')
  creds instance starts: 1 (bad=True 0); gaps s: []
  triggers.failing (latest per kind): asset_watcher=0 callback=0 task=0 
  triggers.failed lines by tag:  41 triggers.failed,trigger_kind=asset_watcher; 1 triggers.failed,trigger_kind=task;
=== C) [09:39:51] fails again, then credentials fixed with no event: recovery by staying up
--- [09:40:03] failing again: row: failures=7  gauge(asset_watcher)=1
--- row:
(1, 8, '09:39:54', '09:40:08', '09:39:39', 'ConnectionError: TO creds: the security token included in the request is expired')
[09:40:10] credentials fixed, no event
[09:40:12] re-created with good credentials, up from about 2026-10-29T09:40:12.039683
--- [09:41:12] up ~60 s, no event yet: row: failures=9  gauge(asset_watcher)=1
--- [09:42:15] up ~120 s: row: failures=None  gauge(asset_watcher)=0
--- row:
(1, None, None, '09:40:10', '09:39:39', 'ConnectionError: TO creds: the security token included in the request is expired')
--- triggerer output: recovery lines
2026-10-29T09:39:39.465335Z [info     ] Trigger recovered              [airflow.jobs.triggerer_job_runner] after_failures=41 loc=triggerer_job_runner.py:961 trigger_id=1 trigger_kind=asset_watcher
2026-10-29T09:42:11.787707Z [info     ] Trigger recovered              [airflow.jobs.triggerer_job_runner] after_failures=9 loc=triggerer_job_runner.py:961 trigger_id=1 trigger_kind=asset_watcher

---

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

• Yes (please specify the tool below)

Generated-by: Claude Code (Opus 5.5) following the guidelines

[potiuk]

Thanks — this reads well. I checked the tracker's recovery and hand-over logic, the triggerer_id-guarded row writes, the masking of last_error (connections and variables fetched through the supervisor are registered with its masker, so redact() covers them), the NUL/surrogate escaping, and the migration's backfill and downgrade. The remaining eager-loading nit only adds a lazy load for rows without an owner during a rolling upgrade, as trigger.assets was already loaded per non-task trigger before, so it doesn't need to block this.

---

Drafted-by: Claude Code (Claude Opus 5); reviewed by @potiuk before posting