From a1ab0653121829675ad864080d2cd9d3d1688b01 Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Wed, 29 Jul 2026 15:53:23 +0200 Subject: [PATCH 1/4] Fail task instances whose stored next_kwargs cannot be processed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit handle_event_submit decoded a task instance's stored next_kwargs and assumed the result was a dict. Neither assumption held: the decode caught only four exception types, so anything the BaseSerialization fallback raised escaped, and the isinstance check sat under TYPE_CHECKING, so it never ran at runtime. Both escape as exceptions from a function whose callers walk every waiting task instance in one pass — the scheduler's timeout sweep and two API routes — so a single unusable payload aborted the whole batch. Decode through a helper that checks its result, and guard decode, event insertion and re-encode together. A task instance whose payload cannot be processed is re-queued to fail through the existing __fail__ path, so its normal retry and callback handling still runs, instead of being left parked for the next sweep to trip over again. Generated-by: Claude Opus 5 (1M context) following the guidelines at https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions --- airflow-core/src/airflow/models/trigger.py | 87 +++++++++++++++---- .../tests/unit/jobs/test_scheduler_job.py | 54 ++++++++++++ .../tests/unit/models/test_trigger.py | 40 +++++++++ 3 files changed, 163 insertions(+), 18 deletions(-) diff --git a/airflow-core/src/airflow/models/trigger.py b/airflow-core/src/airflow/models/trigger.py index 7808d7efa6cbe..785e1f70005c2 100644 --- a/airflow-core/src/airflow/models/trigger.py +++ b/airflow-core/src/airflow/models/trigger.py @@ -500,6 +500,51 @@ def get_sorted_triggers( return result +def _decode_next_kwargs(next_kwargs_raw: Any) -> dict[str, Any]: + """ + Decode the stored ``next_kwargs`` of a task instance into a plain dict. + + Deserialize with serde first to provide a compat layer if there are mixed serialized + (BaseSerialisation and serde) data, which can happen if a deferred task resumes after upgrade. + + The result is checked here rather than assumed, so callers never have to trust the shape of + what comes back out of the stored payload. + + :raise ValueError: The payload did not decode to a dict. + """ + from airflow.sdk.serde import deserialize + + try: + next_kwargs = deserialize(next_kwargs_raw) + except (ImportError, KeyError, AttributeError, TypeError): + from airflow.serialization.serialized_objects import BaseSerialization + + next_kwargs = BaseSerialization.deserialize(next_kwargs_raw) + + if not isinstance(next_kwargs, dict): + raise ValueError(f"next_kwargs decoded to {type(next_kwargs).__name__}, expected a dict") + return next_kwargs + + +def _fail_unresumable_task_instance(task_instance: TaskInstance, reason: str, *, session: Session) -> None: + """ + Re-queue a task instance that cannot be resumed, so that a worker fails it. + + Mirrors :meth:`Trigger.submit_failure`: the special ``__fail__`` next_method makes the worker + fail the task immediately, which runs its normal failure handling (retries and callbacks + included). Leaving the task instance parked instead would strand it there, as the event that + should have resumed it is already gone. + """ + task_instance.next_method = TRIGGER_FAIL_REPR + task_instance.next_kwargs = {"error": reason} + # Remove ourselves as its trigger + task_instance.trigger_id = None + # Finally, mark it as scheduled so it gets re-queued + task_instance.state = TaskInstanceState.SCHEDULED + task_instance.scheduled_dttm = timezone.utcnow() + session.flush() + + @singledispatch def handle_event_submit(event: TriggerEvent, *, task_instance: TaskInstance, session: Session) -> None: """ @@ -509,33 +554,39 @@ def handle_event_submit(event: TriggerEvent, *, task_instance: TaskInstance, ses as well as its state to scheduled. It also adds the event's payload into the kwargs for the task. + A task instance whose stored kwargs cannot be decoded (or re-encoded with the payload added) + is failed rather than resumed. That work happens in the scheduler and API processes, which + handle every parked task instance in turn, so a single unusable payload must not be able to + abort the caller. + :param task_instance: The task instance to handle the submit event for. :param session: The session to be used for the database callback sink. """ - from airflow.sdk.serde import deserialize, serialize + from airflow.sdk.serde import serialize from airflow.utils.state import TaskInstanceState next_kwargs_raw = task_instance.next_kwargs or {} - # deserialize first to provide a compat layer if there are mixed serialized (BaseSerialisation and serde) data - # which can happen if a deferred task resumes after upgrade try: - next_kwargs = deserialize(next_kwargs_raw) - except (ImportError, KeyError, AttributeError, TypeError): - from airflow.serialization.serialized_objects import BaseSerialization - - next_kwargs = BaseSerialization.deserialize(next_kwargs_raw) - - # Add event to the plain dict, then serialize everything together so nested - # non-primitive values get proper serde encoding. - if TYPE_CHECKING: - assert isinstance(next_kwargs, dict) - next_kwargs["event"] = event.payload + next_kwargs = _decode_next_kwargs(next_kwargs_raw) + # Add event to the plain dict, then serialize everything together so nested + # non-primitive values get proper serde encoding. + next_kwargs["event"] = event.payload + # Re-serialize using serde. The Execution API version converter + # (ModifyDeferredTaskKwargsToJsonValue) handles converting this to + # BaseSerialization format when serving old workers. + serialized_next_kwargs = serialize(next_kwargs) + except Exception as exc: + log.exception("Could not process next_kwargs of %s; failing it instead of resuming it", task_instance) + _fail_unresumable_task_instance( + task_instance, + "Could not resume the task: its stored next_kwargs could not be processed " + f"({type(exc).__name__}: {exc})", + session=session, + ) + return - # Re-serialize using serde. The Execution API version converter - # (ModifyDeferredTaskKwargsToJsonValue) handles converting this to - # BaseSerialization format when serving old workers. - task_instance.next_kwargs = serialize(next_kwargs) + task_instance.next_kwargs = serialized_next_kwargs # Remove ourselves as its trigger task_instance.trigger_id = None diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py b/airflow-core/tests/unit/jobs/test_scheduler_job.py index c83d9039d8f6e..9ed75f231a406 100644 --- a/airflow-core/tests/unit/jobs/test_scheduler_job.py +++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py @@ -8213,6 +8213,60 @@ def test_awaiting_input_timeout_without_defaults_fails(self, dag_maker): assert ti.next_method == "execute_complete" assert ti.next_kwargs["event"]["error_type"] == "timeout" + def test_awaiting_input_timeout_sweep_survives_unusable_next_kwargs(self, dag_maker): + """ + A parked task whose stored ``next_kwargs`` cannot be turned into a dict is failed by the + sweep instead of aborting it, and the other timed-out tasks in the same batch are still + resolved. The sweep runs on a recurring scheduler timer, so it has to finish the batch. + """ + session = settings.Session() + with dag_maker( + dag_id="test_awaiting_input_bad_kwargs", + start_date=DEFAULT_DATE, + schedule="@once", + session=session, + ): + EmptyOperator(task_id="dummy1") + dr_bad = dag_maker.create_dagrun() + dr_good = dag_maker.create_dagrun( + run_id="good", logical_date=DEFAULT_DATE + datetime.timedelta(seconds=1) + ) + ti_bad = dr_bad.get_task_instance("dummy1", session=session) + ti_good = dr_good.get_task_instance("dummy1", session=session) + for ti in (ti_bad, ti_good): + ti.state = State.AWAITING_INPUT + ti.trigger_timeout = timezone.utcnow() - datetime.timedelta(seconds=60) + ti.next_method = "execute_complete" + ti.next_kwargs = {} + # Stored kwargs that no longer decode into a dict (here: a class name that is not allowed + # for deserialization, which the legacy fallback cannot read either). + ti_bad.next_kwargs = {"__classname__": "not.allowed.Thing", "__version__": 1, "__data__": {}} + session.add( + HITLDetail( + ti_id=ti_good.id, + options=["Approve", "Reject"], + subject="approve?", + defaults=["Approve"], + multiple=False, + params={}, + ) + ) + session.flush() + + self.job_runner = SchedulerJobRunner(job=Job()) + self.job_runner.check_awaiting_input_timeouts(session=session) + + session.refresh(ti_bad) + session.refresh(ti_good) + # The unreadable one is re-queued for a worker to fail... + assert ti_bad.state == State.SCHEDULED + assert ti_bad.next_method == "__fail__" + assert "error" in ti_bad.next_kwargs + # ...and the healthy one in the same batch is still resumed with its default response. + assert ti_good.state == State.SCHEDULED + assert ti_good.next_method == "execute_complete" + assert ti_good.next_kwargs["event"]["chosen_options"] == ["Approve"] + def test_retry_on_db_error_when_update_timeout_triggers(self, dag_maker, testing_dag_bundle, session): """ Tests that it will retry on DB error like deadlock when updating timeout triggers. diff --git a/airflow-core/tests/unit/models/test_trigger.py b/airflow-core/tests/unit/models/test_trigger.py index 221ffeaa8dc0f..e5d4c673a040d 100644 --- a/airflow-core/tests/unit/models/test_trigger.py +++ b/airflow-core/tests/unit/models/test_trigger.py @@ -35,6 +35,7 @@ from airflow.models.asset import AssetEvent, AssetModel, AssetWatcherModel from airflow.models.callback import Callback, TriggererCallback from airflow.models.taskinstancehistory import TaskInstanceHistory +from airflow.models.trigger import handle_event_submit from airflow.models.xcom import XComModel from airflow.providers.standard.operators.empty import EmptyOperator from airflow.sdk.definitions.callback import AsyncCallback @@ -257,6 +258,45 @@ def test_submit_event_no_n_plus_one_for_assets(_, session, asset_count, expected Trigger.submit_event(trigger_id, TriggerEvent("payload"), session=session) +@pytest.mark.parametrize( + "stored_next_kwargs", + [ + # Decoding blows up: serde rejects the class name, and the BaseSerialization fallback then + # trips over the missing legacy keys. + pytest.param( + {"__classname__": "not.allowed.Thing", "__version__": 1, "__data__": {}}, + id="undecodable", + ), + # Decodes cleanly, but not into a dict: legacy encoding of a bare datetime. + pytest.param({"__type": "datetime", "__var": 1735689600.0}, id="not-a-dict"), + ], +) +def test_handle_event_submit_fails_task_with_unusable_next_kwargs( + session, create_task_instance, stored_next_kwargs +): + """ + Tests that stored kwargs which cannot be turned into a dict fail the task instance instead of + raising out of ``handle_event_submit``. Its callers walk every waiting task instance in one + pass, so one unusable payload must not abort them. + """ + task_instance = create_task_instance( + session=session, logical_date=timezone.utcnow(), state=State.DEFERRED + ) + task_instance.next_method = "execute_complete" + task_instance.next_kwargs = stored_next_kwargs + session.flush() + + handle_event_submit(TriggerEvent("payload"), task_instance=task_instance, session=session) + + session.refresh(task_instance) + # Re-queued for a worker to fail, rather than resumed with kwargs we could not read. + assert task_instance.state == State.SCHEDULED + assert task_instance.next_method == "__fail__" + assert task_instance.trigger_id is None + assert "event" not in task_instance.next_kwargs + assert "next_kwargs could not be processed" in task_instance.next_kwargs["error"] + + def test_submit_failure(session, create_task_instance): """ Tests that failures submitted to a trigger fail their dependent From 8d19b2fea98cc2061b6ecabbf19896814fade6e3 Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Tue, 4 Aug 2026 01:27:12 +0200 Subject: [PATCH 2/4] Report why a task instance could not be resumed, and to whom The Dag author sees only the task log, so the traceback has to travel in next_kwargs the way submit_failure already sends it; the process log where this runs is often not theirs to read. Decode and re-encode also fail for different reasons: blaming the stored kwargs for a payload the trigger just yielded points the author at database state that was never at fault. The sweep's summary counted an unresumable task as resolved. --- .../src/airflow/jobs/scheduler_job_runner.py | 21 ++++-- airflow-core/src/airflow/models/trigger.py | 64 +++++++++++++++---- .../tests/unit/jobs/test_scheduler_job.py | 14 ++-- .../tests/unit/models/test_trigger.py | 31 ++++++++- 4 files changed, 102 insertions(+), 28 deletions(-) diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py b/airflow-core/src/airflow/jobs/scheduler_job_runner.py index 40d1651505794..c6b555cf238eb 100644 --- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py +++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py @@ -3511,8 +3511,10 @@ def check_awaiting_input_timeouts( num_resolved = 0 num_failed = 0 + num_unresumable = 0 for ti in timed_out_tis: hitl_detail = ti.hitl_detail + resuming = True if hitl_detail is not None and hitl_detail.responded_at is not None: # A response landed just before the deadline; resume with it. handle_event_submit( @@ -3520,7 +3522,6 @@ def check_awaiting_input_timeouts( task_instance=ti, session=session, ) - num_resolved += 1 elif hitl_detail is not None and hitl_detail.defaults is not None: # Apply the configured defaults as the response, then resume to success. hitl_detail.chosen_options = list(hitl_detail.defaults) @@ -3536,7 +3537,6 @@ def check_awaiting_input_timeouts( task_instance=ti, session=session, ) - num_resolved += 1 else: # No defaults and no response: resume into execute_complete with a timeout # failure event so the operator raises HITLTimeoutError (matching the old @@ -3552,16 +3552,29 @@ def check_awaiting_input_timeouts( task_instance=ti, session=session, ) + resuming = False + + # ``handle_event_submit`` routes a task instance it could not process to + # ``__fail__`` instead of resuming it. That is neither of the outcomes the + # branches above intended, so it is counted on its own rather than being + # reported as resolved. + if ti.next_method == TRIGGER_FAIL_REPR: + num_unresumable += 1 + elif resuming: + num_resolved += 1 + else: num_failed += 1 # Flush within the retry block so both branches persist consistently (the defaults # branch already flushes via handle_event_submit; the fail branch relies on this). session.flush() - if num_resolved or num_failed: + if num_resolved or num_failed or num_unresumable: self.log.info( - "AWAITING_INPUT timeout sweep: %i resolved (response/defaults), %i failed", + "AWAITING_INPUT timeout sweep: %i resolved (response/defaults), %i failed, " + "%i could not be resumed", num_resolved, num_failed, + num_unresumable, ) # [START find_and_purge_task_instances_without_heartbeats] diff --git a/airflow-core/src/airflow/models/trigger.py b/airflow-core/src/airflow/models/trigger.py index 785e1f70005c2..603e359128783 100644 --- a/airflow-core/src/airflow/models/trigger.py +++ b/airflow-core/src/airflow/models/trigger.py @@ -511,6 +511,8 @@ def _decode_next_kwargs(next_kwargs_raw: Any) -> dict[str, Any]: what comes back out of the stored payload. :raise ValueError: The payload did not decode to a dict. + :raise Exception: Whatever the two decoders raise on a payload they cannot read -- the stored + blob is arbitrary, so the set is open and callers have to treat it as such. """ from airflow.sdk.serde import deserialize @@ -526,7 +528,9 @@ def _decode_next_kwargs(next_kwargs_raw: Any) -> dict[str, Any]: return next_kwargs -def _fail_unresumable_task_instance(task_instance: TaskInstance, reason: str, *, session: Session) -> None: +def _fail_unresumable_task_instance( + task_instance: TaskInstance, reason: str, exc: BaseException, *, session: Session +) -> None: """ Re-queue a task instance that cannot be resumed, so that a worker fails it. @@ -534,12 +538,17 @@ def _fail_unresumable_task_instance(task_instance: TaskInstance, reason: str, *, fail the task immediately, which runs its normal failure handling (retries and callbacks included). Leaving the task instance parked instead would strand it there, as the event that should have resumed it is already gone. + + The traceback travels in ``next_kwargs`` because that is the only channel that reaches the + task log; the exception is otherwise only logged where this runs, which the Dag author may + have no access to. It has to stay the list ``format_exception`` returns -- the runtime joins it. """ task_instance.next_method = TRIGGER_FAIL_REPR - task_instance.next_kwargs = {"error": reason} - # Remove ourselves as its trigger + task_instance.next_kwargs = { + "error": reason, + "traceback": format_exception(type(exc), exc, exc.__traceback__), + } task_instance.trigger_id = None - # Finally, mark it as scheduled so it gets re-queued task_instance.state = TaskInstanceState.SCHEDULED task_instance.scheduled_dttm = timezone.utcnow() session.flush() @@ -554,34 +563,61 @@ def handle_event_submit(event: TriggerEvent, *, task_instance: TaskInstance, ses as well as its state to scheduled. It also adds the event's payload into the kwargs for the task. - A task instance whose stored kwargs cannot be decoded (or re-encoded with the payload added) - is failed rather than resumed. That work happens in the scheduler and API processes, which - handle every parked task instance in turn, so a single unusable payload must not be able to - abort the caller. + A task instance whose stored kwargs cannot be decoded, or which the event payload cannot be + encoded into, is failed rather than resumed. This runs in the triggerer, the scheduler and the + API processes, each of which handles every waiting task instance in one pass, so a single + unusable payload must not be able to abort the caller. The triggerer had the worst of it: an + event whose submit raised was left unconfirmed and redelivered indefinitely. + + Failing the task instance is not free for every caller: a Human-in-the-loop response whose + ``params_input`` serde cannot encode is now recorded and discarded rather than rejected, which + the submitter cannot retry. That wants validating on the write side; tracked at + https://github.com/apache/airflow/issues/71036 :param task_instance: The task instance to handle the submit event for. :param session: The session to be used for the database callback sink. """ from airflow.sdk.serde import serialize - from airflow.utils.state import TaskInstanceState next_kwargs_raw = task_instance.next_kwargs or {} + # Decoding and re-encoding fail for different reasons and are reported separately: blaming the + # stored kwargs for a payload the trigger just yielded would point the author at DB state that + # was never the problem. try: next_kwargs = _decode_next_kwargs(next_kwargs_raw) - # Add event to the plain dict, then serialize everything together so nested - # non-primitive values get proper serde encoding. - next_kwargs["event"] = event.payload + except Exception as exc: + log.exception( + "Could not decode the stored next_kwargs of %s; failing it instead of resuming it", + task_instance, + ) + _fail_unresumable_task_instance( + task_instance, + "Could not resume the task: its stored next_kwargs could not be decoded " + f"({type(exc).__name__}: {exc})", + exc, + session=session, + ) + return + + # Add event to the plain dict, then serialize everything together so nested + # non-primitive values get proper serde encoding. + next_kwargs["event"] = event.payload + try: # Re-serialize using serde. The Execution API version converter # (ModifyDeferredTaskKwargsToJsonValue) handles converting this to # BaseSerialization format when serving old workers. serialized_next_kwargs = serialize(next_kwargs) except Exception as exc: - log.exception("Could not process next_kwargs of %s; failing it instead of resuming it", task_instance) + log.exception( + "Could not serialize the event payload for %s; failing it instead of resuming it", + task_instance, + ) _fail_unresumable_task_instance( task_instance, - "Could not resume the task: its stored next_kwargs could not be processed " + f"Could not resume the task: the event payload could not be serialized " f"({type(exc).__name__}: {exc})", + exc, session=session, ) return diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py b/airflow-core/tests/unit/jobs/test_scheduler_job.py index 9ed75f231a406..d344df97c200c 100644 --- a/airflow-core/tests/unit/jobs/test_scheduler_job.py +++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py @@ -8214,11 +8214,7 @@ def test_awaiting_input_timeout_without_defaults_fails(self, dag_maker): assert ti.next_kwargs["event"]["error_type"] == "timeout" def test_awaiting_input_timeout_sweep_survives_unusable_next_kwargs(self, dag_maker): - """ - A parked task whose stored ``next_kwargs`` cannot be turned into a dict is failed by the - sweep instead of aborting it, and the other timed-out tasks in the same batch are still - resolved. The sweep runs on a recurring scheduler timer, so it has to finish the batch. - """ + """The sweep finishes its batch even when one task's stored kwargs cannot be read.""" session = settings.Session() with dag_maker( dag_id="test_awaiting_input_bad_kwargs", @@ -8254,18 +8250,20 @@ def test_awaiting_input_timeout_sweep_survives_unusable_next_kwargs(self, dag_ma session.flush() self.job_runner = SchedulerJobRunner(job=Job()) - self.job_runner.check_awaiting_input_timeouts(session=session) + mock_log = mock.MagicMock(spec=logging.Logger) + with mock.patch.object(SchedulerJobRunner, "log", mock_log): + self.job_runner.check_awaiting_input_timeouts(session=session) session.refresh(ti_bad) session.refresh(ti_good) - # The unreadable one is re-queued for a worker to fail... assert ti_bad.state == State.SCHEDULED assert ti_bad.next_method == "__fail__" assert "error" in ti_bad.next_kwargs - # ...and the healthy one in the same batch is still resumed with its default response. assert ti_good.state == State.SCHEDULED assert ti_good.next_method == "execute_complete" assert ti_good.next_kwargs["event"]["chosen_options"] == ["Approve"] + # The one it could not resume is reported as such rather than counted as resolved. + assert mock_log.info.call_args.args[1:] == (1, 0, 1) def test_retry_on_db_error_when_update_timeout_triggers(self, dag_maker, testing_dag_bundle, session): """ diff --git a/airflow-core/tests/unit/models/test_trigger.py b/airflow-core/tests/unit/models/test_trigger.py index e5d4c673a040d..b48f00622fe9b 100644 --- a/airflow-core/tests/unit/models/test_trigger.py +++ b/airflow-core/tests/unit/models/test_trigger.py @@ -289,12 +289,39 @@ def test_handle_event_submit_fails_task_with_unusable_next_kwargs( handle_event_submit(TriggerEvent("payload"), task_instance=task_instance, session=session) session.refresh(task_instance) - # Re-queued for a worker to fail, rather than resumed with kwargs we could not read. assert task_instance.state == State.SCHEDULED assert task_instance.next_method == "__fail__" assert task_instance.trigger_id is None assert "event" not in task_instance.next_kwargs - assert "next_kwargs could not be processed" in task_instance.next_kwargs["error"] + assert "stored next_kwargs could not be decoded" in task_instance.next_kwargs["error"] + # The traceback reaches the task log only through next_kwargs, and the runtime joins the list. + assert isinstance(task_instance.next_kwargs["traceback"], list) + + +def test_handle_event_submit_fails_task_when_the_event_payload_cannot_be_serialized( + session, create_task_instance +): + """A payload serde cannot encode is reported as such, not as unreadable stored kwargs. + + Blaming the stored kwargs would point the Dag author at database state that was never the + problem. + """ + task_instance = create_task_instance( + session=session, logical_date=timezone.utcnow(), state=State.DEFERRED + ) + task_instance.next_method = "execute_complete" + task_instance.next_kwargs = {} + session.flush() + + # serde refuses any dict carrying its reserved keys, at any depth. + handle_event_submit( + TriggerEvent({"__classname__": "anything"}), task_instance=task_instance, session=session + ) + + session.refresh(task_instance) + assert task_instance.state == State.SCHEDULED + assert task_instance.next_method == "__fail__" + assert "event payload could not be serialized" in task_instance.next_kwargs["error"] def test_submit_failure(session, create_task_instance): From bf0c0a025dbd142cfa3064c5a5affd6c3cf30177 Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Wed, 5 Aug 2026 04:24:17 +0200 Subject: [PATCH 3/4] Update airflow-core/src/airflow/models/trigger.py Co-authored-by: Amogh Desai --- airflow-core/src/airflow/models/trigger.py | 13 ++++--------- 1 file changed, 4 insertions(+), 9 deletions(-) diff --git a/airflow-core/src/airflow/models/trigger.py b/airflow-core/src/airflow/models/trigger.py index 603e359128783..16d21d91de27a 100644 --- a/airflow-core/src/airflow/models/trigger.py +++ b/airflow-core/src/airflow/models/trigger.py @@ -532,16 +532,11 @@ def _fail_unresumable_task_instance( task_instance: TaskInstance, reason: str, exc: BaseException, *, session: Session ) -> None: """ - Re-queue a task instance that cannot be resumed, so that a worker fails it. + Route through ``__fail__`` (mirrors `Trigger.submit_failure`) so a worker fails the + task normally instead of leaving it stranded with no event left to resume it. - Mirrors :meth:`Trigger.submit_failure`: the special ``__fail__`` next_method makes the worker - fail the task immediately, which runs its normal failure handling (retries and callbacks - included). Leaving the task instance parked instead would strand it there, as the event that - should have resumed it is already gone. - - The traceback travels in ``next_kwargs`` because that is the only channel that reaches the - task log; the exception is otherwise only logged where this runs, which the Dag author may - have no access to. It has to stay the list ``format_exception`` returns -- the runtime joins it. + Traceback goes into ``next_kwargs`` as a list -- the only channel reaching the task log -- + since ``format_exception`` returns it that way and the runtime joins it. """ task_instance.next_method = TRIGGER_FAIL_REPR task_instance.next_kwargs = { From 738251c2e6fe3c89294159d312059cceed8c3e31 Mon Sep 17 00:00:00 2001 From: Rahul Vats Date: Wed, 5 Aug 2026 12:46:38 +0530 Subject: [PATCH 4/4] Fix ruff D205/D213 on _fail_unresumable_task_instance docstring One-line summary on the second line + blank line before the description, so the docstring satisfies both D205 (blank between summary and description) and D213 (summary on second line). Static checks were failing on ruff for this. --- airflow-core/src/airflow/models/trigger.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/airflow-core/src/airflow/models/trigger.py b/airflow-core/src/airflow/models/trigger.py index 16d21d91de27a..a49d4bba2795f 100644 --- a/airflow-core/src/airflow/models/trigger.py +++ b/airflow-core/src/airflow/models/trigger.py @@ -532,9 +532,9 @@ def _fail_unresumable_task_instance( task_instance: TaskInstance, reason: str, exc: BaseException, *, session: Session ) -> None: """ - Route through ``__fail__`` (mirrors `Trigger.submit_failure`) so a worker fails the - task normally instead of leaving it stranded with no event left to resume it. + Route through ``__fail__`` so a worker fails the task normally instead of stranding it. + Mirrors ``Trigger.submit_failure``: without this the task is left with no event to resume it. Traceback goes into ``next_kwargs`` as a list -- the only channel reaching the task log -- since ``format_exception`` returns it that way and the runtime joins it. """