Detect supervisor-subprocess trigger count mismatch in Triggerer - #68792
Detect supervisor-subprocess trigger count mismatch in Triggerer#68792JH0917 wants to merge 3 commits into
Conversation
|
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
|
2a820f1 to
8c3adff
Compare
|
@dstandish @hussein-awala Could you approve the CI run? Thanks! |
|
@potiuk @dstandish @hussein-awala Friendly reminder - could someone approve the CI run? Thanks! |
potiuk
added
the
ready for maintainer review
57f0d9b to
295071e
Compare
|
Force-pushed with a fix for a false-positive path in the mismatch check. |
potiuk
left a comment
potiuk
left a comment
There was a problem hiding this comment.
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
| expected, | ||
| num_running, | ||
| ) | ||
| self.stop = True |
There was a problem hiding this comment.
This exits the supervisor loop via should_stop(), terminating the Triggerer and every trigger it is running. The tasks get picked up by another Triggerer eventually, so it's recoverable — but it's a disruptive, whole-process response to what may be a transient accounting difference.
The docstring concedes the invariant is only valid "between finished-removal and to_create-addition", which is a timing-dependent window: num_running is computed in the subprocess when it builds the message, and compared against supervisor state some time later. I'd want to see the argument for why no legitimate interleaving can produce a one-off difference before wiring it to a shutdown.
Softer options that still surface the bug: log at ERROR and emit a metric, or require N consecutive mismatches before stopping so a single transient blip doesn't cost an outage. If shutdown really is the right call, a comment explaining why the window is airtight would help the next reader.
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
There was a problem hiding this comment.
Thanks -- the shutdown is gone. The check now recovers instead: dropping the stranded ids from running_triggers / cancelling_triggers makes them fall out of known_trigger_ids in update_triggers, so the next loop rebuilds their workloads and nothing else running is disturbed. I added the ERROR log and the triggers.state_mismatch counter you suggested.
On the timing window: both sides only change at the message boundary, and the only code that adds a trigger runs in the runner's main loop -- which is blocked waiting for the reply while the supervisor compares. So the snapshot can't go stale and one disagreement is enough. That's why I skipped the N-consecutive debounce.
Drafted-by: Claude Code (Opus 5); reviewed by @JH0917 before posting
| """ | ||
| expected = len(self.running_triggers) | ||
| if expected != num_running: | ||
| log.error( |
There was a problem hiding this comment.
Uses the module-level log, but _handle_request already receives a bound log: FilteringBoundLogger that carries the request context. Passing that through would keep this message correlated with the rest of the request's logging.
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
There was a problem hiding this comment.
As far as I can see that log only carries logger_name="supervisor" -- nothing is bound to it anywhere on the path. So I left it as is: request context rides on contextvars and shows up either way, and the difference would be that this line gets tagged [supervisor] while the rest of the class logs as [airflow.jobs.triggerer_job_runner]. Happy to be told I'm misreading it.
Drafted-by: Claude Code (Opus 5); reviewed by @JH0917 before posting
| tb = format_exception(type(exc), exc, exc.__traceback__) if exc else None | ||
| failures_to_send.append((trigger_id, tb)) | ||
| if trigger_id not in self.triggers: | ||
| finished_ids.append(trigger_id) |
There was a problem hiding this comment.
Mutating the caller's list in place. I traced it and it's harmless today — finished_ids is freshly built in cleanup_finished_triggers() and isn't read again after sync_state_to_supervisor returns — so this is style, not a bug.
Still, process_trigger_events reads as a pure "build the message" function, and a caller that later reused its list would get a surprise. Building the combined list locally and leaving the parameter untouched keeps that property.
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
There was a problem hiding this comment.
Fixed -- builds a local list now, parameter is untouched, added regression test too. Thanks for pointing that out.
Remove the incorrect difference_update on msg.failures (which would break for serialization failures where the trigger is still running). Instead, include creation-failure IDs in finished_ids so the supervisor naturally removes them from running_triggers via the existing finished processing path. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
295071e to
d0f5f2d
Compare
eladkal
added
type:bug-fix
3 participants
In rare cases the Triggerer subprocess can lose track of a trigger — it
is assigned in the supervisor's
running_triggersbut no coroutine isactually running. The affected task instance stays in
deferredpermanently.
Changes:
Supervisor compares
len(running_triggers)against the subprocess'sactual trigger count each sync cycle. If they diverge, it shuts down
so the orchestrator can restart and recover stuck tasks.
Creation failures are included in
finished_idsso the supervisorremoves them from
running_triggersnaturally (prevents false-positiveshutdown).
related: #63913
Was generative AI tooling used to co-author this PR?
Generated-by: Claude Code (Opus 4.6) following the guidelines