Detect supervisor-subprocess trigger count mismatch in Triggerer

airflow-core/newsfragments/68792.bugfix.rst

@@ -0,0 +1 @@
+Recover Triggerer triggers that the supervisor tracked as running but for which the runner had no live coroutine. Their task instances previously stayed in ``deferred`` indefinitely; the trigger is now re-created on the next sync so the task can resume. A new ``triggers.state_mismatch`` counter is emitted when this is detected.

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

@@ -445,6 +445,8 @@ class TriggerStateChanges(BaseModel):
         # Format of list[str] is the exc traceback format
         failures: list[tuple[int, list[str] | None]] | None = None
         finished: list[int] | None = None
+        # Ids the runner has a live coroutine for; None when the message carries no information
+        running_ids: set[int] | None = None
 
     class TriggerStateSync(BaseModel):
         type: Literal["TriggerStateSync"] = "TriggerStateSync"
@@ -748,6 +750,8 @@ def _handle_request(self, msg: ToTriggerSupervisor, log: FilteringBoundLogger, r
                         # handle leaks for every failed upload.
                         factory.close()
 
+            self.check_for_unhandled_triggers(msg.running_ids)
+
             # Drain the persist confirmations accumulated since the last sync.
             events_persisted: list[int] = []
             while self.persisted_event_seqs:
@@ -933,6 +937,24 @@ def clean_unused(self) -> None:
             with attempt, create_session() as session:
                 Trigger.clean_unused(session=session)
 
+    def check_for_unhandled_triggers(self, running_ids: set[int] | None) -> None:
+        """
+        Re-create triggers we track as running that the runner has no coroutine for.
+
+        Only valid between finished-removal and to_create-addition in ``_handle_request``, where the
+        two sides agree. Dropping the leftovers lets :meth:`update_triggers` rebuild them.
+        """
+        if running_ids is None:
+            return
+        unhandled = self.running_triggers - running_ids
+        if not unhandled:
+            return
+        log.error("Triggers have no coroutine in the runner; re-creating", trigger_ids=sorted(unhandled))
+        self.running_triggers -= unhandled
+        self.cancelling_triggers -= unhandled
+        self.releasing_triggers -= unhandled
+        stats.incr("triggers.state_mismatch", len(unhandled), tags=prune_dict({"team_name": self.team_name}))
+
     def handle_failed_triggers(self):
         """
         Handle "failed" triggers. - ones that errored or exited before they sent an event.
@@ -1805,6 +1827,7 @@ def process_trigger_events(self, finished_ids: list[int]) -> messages.TriggerSta
         # Copy out of our dequeues in threadsafe manner to sync state with parent
         events_to_send: list[TriggerEventEntry] = []
         failures_to_send: list[tuple[int, list[str] | None]] = []
+        finished_to_send = list(finished_ids)
 
         while self.events:
             events_to_send.append(self.events.popleft())
@@ -1813,11 +1836,14 @@ def process_trigger_events(self, finished_ids: list[int]) -> messages.TriggerSta
             trigger_id, exc = self.failed_triggers.popleft()
             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 and trigger_id not in finished_to_send:
+                finished_to_send.append(trigger_id)
 
         return messages.TriggerStateChanges(
             events=events_to_send if events_to_send else None,
-            finished=finished_ids if finished_ids else None,
+            finished=finished_to_send if finished_to_send else None,
             failures=failures_to_send if failures_to_send else None,
+            running_ids=set(self.triggers),
         )
 
     def sanitize_trigger_events(self, msg: messages.TriggerStateChanges) -> messages.TriggerStateChanges:
@@ -1846,6 +1872,7 @@ def sanitize_trigger_events(self, msg: messages.TriggerStateChanges) -> messages
             events=events_to_send if events_to_send else None,
             finished=msg.finished,
             failures=msg.failures,
+            running_ids=msg.running_ids,
         )
 
     async def sync_state_to_supervisor(self, finished_ids: list[int]) -> None:

airflow-core/tests/unit/jobs/test_triggerer_job.py

@@ -4478,3 +4478,136 @@ def test_a_recorded_owner_is_kept(self, mocker):
         TriggerRunnerSupervisor._record_missing_owner(trigger)
 
         assert (trigger.dag_id, trigger.owner_ref) == ("d", {"ti_id": "x"})
+
+
+class TestCheckForUnhandledTriggers:
+    """Tests for recovery of triggers the runner has no coroutine for."""
+
+    @pytest.mark.parametrize(
+        ("running_triggers", "running_ids", "expected_after"),
+        [
+            pytest.param({1, 2, 3}, {1, 2, 3}, {1, 2, 3}, id="in-sync"),
+            pytest.param(set(), set(), set(), id="no-triggers"),
+            pytest.param({1, 2, 3}, {1, 2}, {1, 2}, id="one-unhandled"),
+            pytest.param({1, 2, 3}, set(), set(), id="all-unhandled"),
+        ],
+    )
+    def test_unhandled_triggers_are_dropped(
+        self, jobless_supervisor, mocker, running_triggers, running_ids, expected_after
+    ):
+        incr = mocker.patch("airflow.jobs.triggerer_job_runner.stats.incr", autospec=True)
+        jobless_supervisor.running_triggers = set(running_triggers)
+        jobless_supervisor.cancelling_triggers = set(running_triggers)
+
+        jobless_supervisor.check_for_unhandled_triggers(running_ids)
+
+        unhandled = running_triggers - running_ids
+        assert jobless_supervisor.running_triggers == expected_after
+        assert jobless_supervisor.cancelling_triggers == expected_after
+        assert (
+            mocker.call("triggers.state_mismatch", len(unhandled), tags={}) in incr.call_args_list
+        ) is bool(unhandled)
+
+    def test_dropped_trigger_is_recreated_next_loop(self, jobless_supervisor, mocker):
+        """Dropping an unhandled id is what lets update_triggers rebuild its workload."""
+        build = mocker.patch.object(
+            TriggerRunnerSupervisor, "build_trigger_workloads", autospec=True, return_value=[]
+        )
+        jobless_supervisor.running_triggers = {1, 2, 3}
+        # A trigger released to another triggerer stays in running_triggers, so it can turn up
+        # unhandled too, and releasing_triggers also gates re-creation via known_trigger_ids.
+        jobless_supervisor.releasing_triggers = {3}
+
+        jobless_supervisor.check_for_unhandled_triggers({1})
+        jobless_supervisor.update_triggers({1, 2, 3})
+
+        assert build.call_args.args[1] == {2, 3}
+
+    def test_message_without_running_ids_is_ignored(self, jobless_supervisor):
+        """``None`` means the message carried no information, which must not read as "nothing running"."""
+        jobless_supervisor.running_triggers = {1, 2, 3}
+        jobless_supervisor.cancelling_triggers = {1, 2, 3}
+
+        jobless_supervisor.check_for_unhandled_triggers(None)
+
+        assert jobless_supervisor.running_triggers == {1, 2, 3}
+        assert jobless_supervisor.cancelling_triggers == {1, 2, 3}
+
+    def test_handle_request_checks_before_adding_to_create(self, jobless_supervisor, mocker):
+        """The check fires after finished processing but before to_create is added to running_triggers."""
+        mocker.patch.object(TriggerRunnerSupervisor, "send_msg", autospec=True)
+        jobless_supervisor.running_triggers = {1, 2}
+        jobless_supervisor.creating_triggers.append(
+            mocker.MagicMock(id=3),
+        )
+
+        jobless_supervisor._handle_request(
+            messages.TriggerStateChanges(
+                events=None,
+                failures=None,
+                finished=[1],
+                running_ids={2},
+            ),
+            log=MagicMock(spec=FilteringBoundLogger),
+            req_id=1,
+        )
+
+        assert jobless_supervisor.running_triggers == {2, 3}
+
+    def test_creation_failure_reported_in_finished(self, jobless_supervisor, mocker):
+        """A creation failure appears in both failures and finished, so running_triggers stays in sync."""
+        mocker.patch.object(TriggerRunnerSupervisor, "send_msg", autospec=True)
+        jobless_supervisor.running_triggers = {1, 2, 3}
+
+        jobless_supervisor._handle_request(
+            messages.TriggerStateChanges(
+                events=None,
+                failures=[(3, ["Traceback..."])],
+                finished=[3],
+                running_ids={1, 2},
+            ),
+            log=MagicMock(spec=FilteringBoundLogger),
+            req_id=1,
+        )
+
+        assert 3 not in jobless_supervisor.running_triggers
+
+
+class TestCreationFailureInFinished:
+    """Tests that creation failures (not in self.triggers) are included in finished_ids."""
+
+    def test_creation_failure_included_in_finished(self):
+        runner = TriggerRunner()
+        runner.failed_triggers.append((42, ValueError("bad classpath")))
+
+        msg = runner.process_trigger_events(finished_ids=[10])
+
+        assert msg.finished == [10, 42]
+        assert msg.running_ids == set()
+
+    def test_serialization_failure_not_in_finished(self):
+        runner = TriggerRunner()
+        runner.triggers = {42: {"task": MagicMock(), "is_watcher": False, "name": "t", "events": 0}}
+        runner.failed_triggers.append((42, ValueError("not serializable")))
+
+        msg = runner.process_trigger_events(finished_ids=[])
+
+        assert msg.finished is None
+        assert msg.running_ids == {42}
+
+    def test_failure_already_in_finished_ids_not_duplicated(self):
+        runner = TriggerRunner()
+        runner.failed_triggers.append((42, ValueError("exited without an event")))
+
+        msg = runner.process_trigger_events(finished_ids=[42])
+
+        assert msg.finished == [42]
+
+    def test_caller_finished_ids_not_mutated(self):
+        runner = TriggerRunner()
+        runner.failed_triggers.append((42, ValueError("bad classpath")))
+        finished_ids = [10]
+
+        runner.process_trigger_events(finished_ids=finished_ids)
+
+        assert finished_ids == [10]

shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml

@@ -378,6 +378,13 @@ metrics:
     legacy_name: "-"
     name_variables: []
 
+  - name: "triggers.state_mismatch"
+    description: "Number of triggers the Triggerer tracked as running that the trigger runner
+    had no coroutine for, and which were re-created"
+    type: "counter"
+    legacy_name: "-"
+    name_variables: []
+
   - name: "triggers.succeeded"
     description: "Number of triggers that have fired at least one event.
     Tagged with trigger_kind (asset_watcher, callback or task)."