[JH0917]

In rare cases the Triggerer subprocess can lose track of a trigger — it
is assigned in the supervisor's running_triggers but no coroutine is
actually running. The affected task instance stays in deferred
permanently.

Changes:

1. Supervisor compares len(running_triggers) against the subprocess's
   actual trigger count each sync cycle. If they diverge, it shuts down
   so the orchestrator can restart and recover stuck tasks.

2. Creation failures are included in finished_ids so the supervisor
   removes them from running_triggers naturally (prevents false-positive
   shutdown).

related: #63913

---

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

• Yes — Claude Code (Opus 4.6)

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

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example Dag that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[JH0917]

@dstandish @hussein-awala Could you approve the CI run? Thanks!

[JH0917]

@potiuk @dstandish @hussein-awala Friendly reminder - could someone approve the CI run? Thanks!

[JH0917]

Force-pushed with a fix for a false-positive path in the mismatch check.

[potiuk]

Thanks — detecting supervisor/subprocess drift is worth doing, and I checked the placement: the call sits between running_triggers.discard(id) for finished triggers (line 594) and running_triggers.update(...) for to_create (line 621), which is exactly the window your docstring describes. Test coverage is good too.

What I'd push back on is the remedy. self.stop = True feeds should_stop() and exits the supervisor run loop, so a single count disagreement takes down the whole Triggerer and every trigger running on it. That's a very large hammer for a condition the code itself describes as only valid inside a narrow window. Detail inline.

As with my other reviews today: these are design questions more than defects, my review was AI-assisted, and I'd rather you formed your own view than took mine as settled. Other maintainers who know the Triggerer supervisor protocol should weigh in before this lands.

---

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

[JH0917]

@potiuk -- the three points above are addressed and CI is green. The CHANGES_REQUESTED still left, so a re-look when you get a chance would help. Happy to adjust if the reasoning on the timing window doesn't hold up.

@dstandish @hussein-awala -- you're the CODEOWNERS for triggerer_job_runner.py, and this touches the supervisor/runner sync protocol that @potiuk suggested should get a second pair of eyes. Would appreciate a look when you have time.

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

[potiuk]

Everything from the last round is addressed, and one of my points was wrong.

The remedy rework is what I was hoping for. Dropping the stranded ids so update_triggers rebuilds them is recoverable in a way self.stop = True was not, and the ERROR log plus the triggers.state_mismatch counter gives operators something to alert on rather than an unexplained restart.

On the bound logger: you are right and I was wrong. I traced it — _register_pipe_readers takes no log parameter, so the log that reaches _handle_request is the module-level structlog.get_logger(logger_name="supervisor") in supervisor.py. Nothing is bound to it, so there is no request context to preserve. Leave it as it is.

On skipping the debounce: your argument holds as far as I can follow it. running_ids is snapshotted before the await, so a trigger finishing mid-window is still in the snapshot and will not be flagged — the race only errs in the safe direction.

One new thing, inline: the running_ids default.

@hussein-awala — you are CODEOWNER for triggerer_job_runner.py and have been in supervisor.py recently. Could you look at the supervisor/runner protocol reasoning here? The open question is whether a single count disagreement can ever be legitimate, which is what decides whether the no-debounce choice is right, and I am not the right person to settle that.

---

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

[potiuk]

An empty set means "the runner has no live coroutines", which is indistinguishable from "this message carried no running_ids at all". In the second case unhandled = self.running_triggers - set() is everything, so the supervisor drops and re-creates every running trigger.

No production path hits this today — process_trigger_events and sanitize_trigger_events both set the field. But the existing tests construct the message without it (e.g. TriggerStateChanges(events=None, failures=None, finished=None) in test_triggerer_job.py), which is exactly that shape. With this carrying backport-to-v3-3-test, I would rather the ambiguity not exist at all:

Suggested change

	running_ids: set[int] = set()
	# Ids the runner has a live coroutine for; None when the message carries no information
	running_ids: set[int] | None = None

with an early return in check_for_unhandled_triggers when running_ids is None.

---

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

[JH0917]

Applied -- thanks for the review. I agree that if the field is missing, defaulting to None is safer than creating all triggers again. Added a test for the None path.

[potiuk]

This round is good. The running_ids default is what I asked for, and I no longer need the second opinion I asked for last time — I traced it myself.

On the timing window: arun() runs cleanup_finished_triggers() → sync_state_to_supervisor() → create_triggers() → cancel_triggers() in one task, and process_trigger_events snapshots set(self.triggers) before the asend await. So the runner is parked on that await for the whole time the supervisor is comparing, and nothing else in the process mutates self.triggers. The snapshot cannot go stale mid-comparison, which is your argument, and it holds. One disagreement is genuine — no debounce needed.

Also checked that set[int] already crosses this channel (TriggerStateSync.to_cancel), so running_ids needs no new serialization support, and the metrics registry check passes for the new counter. Locally: the 11 new tests pass and the rest of test_triggerer_job.py is unaffected.

Two things to fix before merge:

1. Please add airflow-core/newsfragments/68792.bugfix.rst. The fix is user-visible (tasks that would have stayed deferred now recover) and triggers.state_mismatch is a new public metric operators will alert on — the two sibling Triggerer metric PRs (#67927, #68521) both carry one. With backport-to-v3-3-test on this, the newsfragment is what puts it in the 3.3.x release notes.

2. The duplicate id in finished, inline below.

---

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

[JH0917]

Thanks for tracing the timing window yourself.

Both items are in: the newsfragment, and the duplicate-id guard with a
regression test. I also rebased onto main.

That was a force-push, so the workflow runs need approving again -- could you
kick them off when you get a chance?

---

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

[vatsrahul1001]

Can merge post code owners review

[JH0917]

@potiuk Rebased onto main and resolved the conflict -- an append-at-end-of-file
clash in test_triggerer_job.py with #67244, so both sets of tests are kept.
The workflow runs need approving again.

Should we still wait for a code owner review here? The note above says it is
needed before merge, but you said last time that you no longer needed the
second opinion, and the code owners have not been active on this PR -- so I am
not sure whether this is waiting on something real.

[potiuk]

This one looks good but @hussein-awala -> maybe you could confirm it ?

[potiuk]

One more thing while this waits for the code-owner look: #73454 added a third tracking set, releasing_triggers, after this PR was written, and the new recovery path doesn't clear it.

check_for_unhandled_triggers (triggerer_job_runner.py:811-816) removes the unhandled ids from running_triggers and cancelling_triggers, but not from releasing_triggers. A trigger that was being released to another triggerer (it is added there from running_triggers at :1055) when its coroutine vanished stays in releasing_triggers, and since that set is part of known_trigger_ids (:1021-1027), it is never re-created if it is later assigned back to this triggerer — the same stuck-in-deferred outcome this PR fixes.

Could you add self.releasing_triggers -= unhandled next to the other two, and extend the recovery test so one of the unhandled ids is also in releasing_triggers and assert it gets re-created?

---

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

[JH0917]

Added self.releasing_triggers -= unhandled (:817) and extended the recovery
test to cover an unhandled id that is also in releasing_triggers, asserting
it gets re-created. The workflow runs need approving again.

It traces exactly as you said. A trigger being released is added to
releasing_triggers (:1056) and stays in running_triggers as well
(:1029-1030), so it can be among the unhandled ids. All three sets feed
known_trigger_ids (:1022-1027), so clearing only two of them leaves the id
there and update_triggers never rebuilds it. Besides the line above, the only
thing that clears releasing_triggers is the finished handler at :608, and
the re-sent to_release cannot produce that report -- cancel_triggers skips
ids it does not hold (:1513).

---

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

[seanghaeli]

All concerns have been addressed, @potiuk good to merge?

[JH0917]

Updated the branch and resolved the conflict.

@dstandish @hussein-awala code-owner review needed for airflow-core/src/airflow/jobs/triggerer_job_runner.py.