Load the correct DAG version when a task starts from a trigger

[seanghaeli]

If a Dag run without a bundle_version and one of its tasks point at different Dag versions, the trigger should load the latest version. Currently, it loads the task's possibly outdated version.

[seanghaeli]

@o-nikolas this one is similar in spirit to your PR #69941

[o-nikolas]

Seems like a reasonable change semantically. But someone who knows the trigger code better should review to see if this is what we want (it seems like it should be though to me at least). Maybe @vincbeck?

[viiccwen]

left comments.

[viiccwen]

created_dag_version_id is populated even for Dags with disable_bundle_versioning=True; dag_run.bundle_version is what determines
whether the run is pinned. This unconditional preference for created_dag_version_id therefore makes an unpinned start_from_trigger task load the Dag version from when the run was created, even after the scheduler has advanced the unfinished TI's dag_version_id following a reparse.

This differs from DBDagBag._version_from_dag_run(), which intentionally resolves the latest version when bundle_version is absent. Please use the run's created_dag_version_id only when the run is pinned, retain the TI version for unpinned runs, and cover both cases in the regression test.

[seanghaeli]

I believe this is addressed now, could you take a look?

[o-nikolas]

@viiccwen ☝️

[potiuk]

@seanghaeli — There are 3 unresolved review thread(s) on this PR from @viiccwen. Could you either push a fix or reply in each thread explaining why the feedback doesn't apply? Once you believe the feedback is addressed, mark the thread as resolved so the reviewer isn't re-pinged needlessly. Thanks!

---

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.

[o-nikolas]

@viiccwen do the changes look good to you now?

[vatsrahul1001]

LGTM, I think we are good to merge after code owners review @dstandish@hussein-awala

[jason810496]

Thanks for the fix.

IMO, we should introduce something like get_serialized_dag_model_for_run rather than calling the private method of the DagBag.

# airflow-core/src/airflow/models/dagbag.py, next to get_dag_for_rundefget_serialized_dag_model_for_run(
self, dag_run: DagRun, *, session: Session
) ->SerializedDagModel|None:
"""Return the SerializedDagModel for the version a run executes against."""ifversion_id:=self._version_from_dag_run(dag_run=dag_run, session=session):
returnself.get_serialized_dag_model(version_id=version_id, session=session)
returnNone

airflow/airflow-core/src/airflow/models/dagbag.py

Lines 218 to 221 in b3cc48d

	defget_dag_for_run(self, dag_run: DagRun, session: Session) ->SerializedDAG|None:
	ifversion_id:=self._version_from_dag_run(dag_run=dag_run, session=session):
	returnself._get_dag(version_id=version_id, session=session)
	returnNone

[vatsrahul1001]

Moving to 3.3.2 as this is still pending code owner review and do not want to rush on merging this without that

[vatsrahul1001]

@dstandish@hussein-awala can you review this?

[hussein-awala]

Thanks for the fix, resolving the version from the run is the right direction and it
matches what workers already do. One thing worries me though. This applies to every
trigger, not only the start_from_trigger ones, and that makes the get_task call
able to kill the triggerer.

Here is the scenario. Take a DAG in a bundle without versioning, so its runs have
bundle_version = None.

1. my_dag v1 has a deferrable task wait. A run is created at v1 and wait
   defers. The TI is DEFERRED with dag_version_id = v1 and there is a Trigger
   row for it.
2. Someone edits the file and renames wait to wait_for_data. The dag processor
   creates v2 and it becomes the latest. The run is still RUNNING, its
   bundle_version is still None and created_dag_version_id is still v1.
3. The scheduler does not clean the TI up yet.
   _check_for_removed_or_restored_tasks only marks it REMOVED when
   self.state != DagRunState.RUNNING, so nothing happens until the next
   task_instance_scheduling_decisions pass over that run.
4. In the meantime the triggerer builds the workload for that trigger, which happens
   on a restart, on a rolling deploy, or on HA failover after assign_unassigned.
   _version_from_dag_run sees no bundle_version, returns v2 as the latest, and
   serialized_dag_model.dag.get_task("wait") raises TaskNotFound.
5. Nobody catches it. Not _create_workload, not build_trigger_workloads, not
   run_once. TriggererJobRunner.run() logs it and re-raises, so the process exits
   and every other trigger it was running goes down with it. After the restart
   assign_unassigned gives it the same trigger back and it crashes again, until the
   scheduler gets around to marking the TI REMOVED and clean_unused deletes the
   trigger.

Steps 2 and 4 are really the same event in practice. A rolling deploy of new DAG
code is exactly when new versions show up and triggerers restart.

Before this change the version came from the TI, so the task was always there and
step 4 just worked. We already guard this pattern elsewhere. In
task_instances.py the same resolution is followed by
with contextlib.suppress(TaskNotFound).

The simplest fix I can think of is to keep the flag check on the TI version and only
re-resolve for the case the title is about:

serialized_dag_model=dag_bag.get_serialized_dag_model(
version_id=trigger.task_instance.dag_version_id, session=session
)
ifserialized_dag_model:
task=serialized_dag_model.dag.get_task(trigger.task_instance.task_id)
iftask.start_from_trigger:
dag_run=trigger.task_instance.get_dagrun(session=session)
run_version_id=DBDagBag._version_from_dag_run(dag_run=dag_run, session=session)
ifrun_version_idandrun_version_id!=trigger.task_instance.dag_version_id:
serialized_dag_model= (
dag_bag.get_serialized_dag_model(version_id=run_version_id, session=session)
orserialized_dag_model
)

[seanghaeli]

@hussein-awala good catch, it can crash in this specific sequence. In fact, the current version of main can fail for a similar reason noted in issue #69841. The latest commit introduces a guard so it falls back to running the trigger without Dag context instead of killing the triggerer

[seanghaeli]

If a Dag run without a bundle_version and one of its tasks point at different Dag versions, the trigger should load the latest version. Currently, it loads the task's possibly outdated version.

[seanghaeli]

@o-nikolas this one is similar in spirit to your PR #69941

[o-nikolas]

Seems like a reasonable change semantically. But someone who knows the trigger code better should review to see if this is what we want (it seems like it should be though to me at least). Maybe @vincbeck?

[viiccwen]

left comments.

[viiccwen]

created_dag_version_id is populated even for Dags with disable_bundle_versioning=True; dag_run.bundle_version is what determines
whether the run is pinned. This unconditional preference for created_dag_version_id therefore makes an unpinned start_from_trigger task load the Dag version from when the run was created, even after the scheduler has advanced the unfinished TI's dag_version_id following a reparse.

This differs from DBDagBag._version_from_dag_run(), which intentionally resolves the latest version when bundle_version is absent. Please use the run's created_dag_version_id only when the run is pinned, retain the TI version for unpinned runs, and cover both cases in the regression test.

[seanghaeli]

I believe this is addressed now, could you take a look?

[o-nikolas]

@viiccwen ☝️

[potiuk]

@seanghaeli — There are 3 unresolved review thread(s) on this PR from @viiccwen. Could you either push a fix or reply in each thread explaining why the feedback doesn't apply? Once you believe the feedback is addressed, mark the thread as resolved so the reviewer isn't re-pinged needlessly. Thanks!

---

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.

[o-nikolas]

@viiccwen do the changes look good to you now?

[vatsrahul1001]

LGTM, I think we are good to merge after code owners review @dstandish@hussein-awala

[jason810496]

Thanks for the fix.

IMO, we should introduce something like get_serialized_dag_model_for_run rather than calling the private method of the DagBag.

# airflow-core/src/airflow/models/dagbag.py, next to get_dag_for_rundefget_serialized_dag_model_for_run(
self, dag_run: DagRun, *, session: Session
) ->SerializedDagModel|None:
"""Return the SerializedDagModel for the version a run executes against."""ifversion_id:=self._version_from_dag_run(dag_run=dag_run, session=session):
returnself.get_serialized_dag_model(version_id=version_id, session=session)
returnNone

airflow/airflow-core/src/airflow/models/dagbag.py

Lines 218 to 221 in b3cc48d

	defget_dag_for_run(self, dag_run: DagRun, session: Session) ->SerializedDAG|None:
	ifversion_id:=self._version_from_dag_run(dag_run=dag_run, session=session):
	returnself._get_dag(version_id=version_id, session=session)
	returnNone

[vatsrahul1001]

Moving to 3.3.2 as this is still pending code owner review and do not want to rush on merging this without that

[vatsrahul1001]

@dstandish@hussein-awala can you review this?

[hussein-awala]

Thanks for the fix, resolving the version from the run is the right direction and it
matches what workers already do. One thing worries me though. This applies to every
trigger, not only the start_from_trigger ones, and that makes the get_task call
able to kill the triggerer.

Here is the scenario. Take a DAG in a bundle without versioning, so its runs have
bundle_version = None.

1. my_dag v1 has a deferrable task wait. A run is created at v1 and wait
   defers. The TI is DEFERRED with dag_version_id = v1 and there is a Trigger
   row for it.
2. Someone edits the file and renames wait to wait_for_data. The dag processor
   creates v2 and it becomes the latest. The run is still RUNNING, its
   bundle_version is still None and created_dag_version_id is still v1.
3. The scheduler does not clean the TI up yet.
   _check_for_removed_or_restored_tasks only marks it REMOVED when
   self.state != DagRunState.RUNNING, so nothing happens until the next
   task_instance_scheduling_decisions pass over that run.
4. In the meantime the triggerer builds the workload for that trigger, which happens
   on a restart, on a rolling deploy, or on HA failover after assign_unassigned.
   _version_from_dag_run sees no bundle_version, returns v2 as the latest, and
   serialized_dag_model.dag.get_task("wait") raises TaskNotFound.
5. Nobody catches it. Not _create_workload, not build_trigger_workloads, not
   run_once. TriggererJobRunner.run() logs it and re-raises, so the process exits
   and every other trigger it was running goes down with it. After the restart
   assign_unassigned gives it the same trigger back and it crashes again, until the
   scheduler gets around to marking the TI REMOVED and clean_unused deletes the
   trigger.

Steps 2 and 4 are really the same event in practice. A rolling deploy of new DAG
code is exactly when new versions show up and triggerers restart.

Before this change the version came from the TI, so the task was always there and
step 4 just worked. We already guard this pattern elsewhere. In
task_instances.py the same resolution is followed by
with contextlib.suppress(TaskNotFound).

The simplest fix I can think of is to keep the flag check on the TI version and only
re-resolve for the case the title is about:

serialized_dag_model=dag_bag.get_serialized_dag_model(
version_id=trigger.task_instance.dag_version_id, session=session
)
ifserialized_dag_model:
task=serialized_dag_model.dag.get_task(trigger.task_instance.task_id)
iftask.start_from_trigger:
dag_run=trigger.task_instance.get_dagrun(session=session)
run_version_id=DBDagBag._version_from_dag_run(dag_run=dag_run, session=session)
ifrun_version_idandrun_version_id!=trigger.task_instance.dag_version_id:
serialized_dag_model= (
dag_bag.get_serialized_dag_model(version_id=run_version_id, session=session)
orserialized_dag_model
)

[seanghaeli]

@hussein-awala good catch, it can crash in this specific sequence. In fact, the current version of main can fail for a similar reason noted in issue #69841. The latest commit introduces a guard so it falls back to running the trigger without Dag context instead of killing the triggerer

[seanghaeli]

If a Dag run without a bundle_version and one of its tasks point at different Dag versions, the trigger should load the latest version. Currently, it loads the task's possibly outdated version.

[seanghaeli]

@o-nikolas this one is similar in spirit to your PR #69941

[o-nikolas]

Seems like a reasonable change semantically. But someone who knows the trigger code better should review to see if this is what we want (it seems like it should be though to me at least). Maybe @vincbeck?

[viiccwen]

left comments.

[viiccwen]

created_dag_version_id is populated even for Dags with disable_bundle_versioning=True; dag_run.bundle_version is what determines
whether the run is pinned. This unconditional preference for created_dag_version_id therefore makes an unpinned start_from_trigger task load the Dag version from when the run was created, even after the scheduler has advanced the unfinished TI's dag_version_id following a reparse.

This differs from DBDagBag._version_from_dag_run(), which intentionally resolves the latest version when bundle_version is absent. Please use the run's created_dag_version_id only when the run is pinned, retain the TI version for unpinned runs, and cover both cases in the regression test.

[seanghaeli]

I believe this is addressed now, could you take a look?

[o-nikolas]

@viiccwen ☝️

[potiuk]

@seanghaeli — There are 3 unresolved review thread(s) on this PR from @viiccwen. Could you either push a fix or reply in each thread explaining why the feedback doesn't apply? Once you believe the feedback is addressed, mark the thread as resolved so the reviewer isn't re-pinged needlessly. Thanks!

---

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.

[o-nikolas]

@viiccwen do the changes look good to you now?

[vatsrahul1001]

LGTM, I think we are good to merge after code owners review @dstandish@hussein-awala

[jason810496]

Thanks for the fix.

IMO, we should introduce something like get_serialized_dag_model_for_run rather than calling the private method of the DagBag.

# airflow-core/src/airflow/models/dagbag.py, next to get_dag_for_rundefget_serialized_dag_model_for_run(
self, dag_run: DagRun, *, session: Session
) ->SerializedDagModel|None:
"""Return the SerializedDagModel for the version a run executes against."""ifversion_id:=self._version_from_dag_run(dag_run=dag_run, session=session):
returnself.get_serialized_dag_model(version_id=version_id, session=session)
returnNone

airflow/airflow-core/src/airflow/models/dagbag.py

Lines 218 to 221 in b3cc48d

	defget_dag_for_run(self, dag_run: DagRun, session: Session) ->SerializedDAG|None:
	ifversion_id:=self._version_from_dag_run(dag_run=dag_run, session=session):
	returnself._get_dag(version_id=version_id, session=session)
	returnNone

[vatsrahul1001]

Moving to 3.3.2 as this is still pending code owner review and do not want to rush on merging this without that

[vatsrahul1001]

@dstandish@hussein-awala can you review this?

[hussein-awala]

Thanks for the fix, resolving the version from the run is the right direction and it
matches what workers already do. One thing worries me though. This applies to every
trigger, not only the start_from_trigger ones, and that makes the get_task call
able to kill the triggerer.

Here is the scenario. Take a DAG in a bundle without versioning, so its runs have
bundle_version = None.

1. my_dag v1 has a deferrable task wait. A run is created at v1 and wait
   defers. The TI is DEFERRED with dag_version_id = v1 and there is a Trigger
   row for it.
2. Someone edits the file and renames wait to wait_for_data. The dag processor
   creates v2 and it becomes the latest. The run is still RUNNING, its
   bundle_version is still None and created_dag_version_id is still v1.
3. The scheduler does not clean the TI up yet.
   _check_for_removed_or_restored_tasks only marks it REMOVED when
   self.state != DagRunState.RUNNING, so nothing happens until the next
   task_instance_scheduling_decisions pass over that run.
4. In the meantime the triggerer builds the workload for that trigger, which happens
   on a restart, on a rolling deploy, or on HA failover after assign_unassigned.
   _version_from_dag_run sees no bundle_version, returns v2 as the latest, and
   serialized_dag_model.dag.get_task("wait") raises TaskNotFound.
5. Nobody catches it. Not _create_workload, not build_trigger_workloads, not
   run_once. TriggererJobRunner.run() logs it and re-raises, so the process exits
   and every other trigger it was running goes down with it. After the restart
   assign_unassigned gives it the same trigger back and it crashes again, until the
   scheduler gets around to marking the TI REMOVED and clean_unused deletes the
   trigger.

Steps 2 and 4 are really the same event in practice. A rolling deploy of new DAG
code is exactly when new versions show up and triggerers restart.

Before this change the version came from the TI, so the task was always there and
step 4 just worked. We already guard this pattern elsewhere. In
task_instances.py the same resolution is followed by
with contextlib.suppress(TaskNotFound).

The simplest fix I can think of is to keep the flag check on the TI version and only
re-resolve for the case the title is about:

serialized_dag_model=dag_bag.get_serialized_dag_model(
version_id=trigger.task_instance.dag_version_id, session=session
)
ifserialized_dag_model:
task=serialized_dag_model.dag.get_task(trigger.task_instance.task_id)
iftask.start_from_trigger:
dag_run=trigger.task_instance.get_dagrun(session=session)
run_version_id=DBDagBag._version_from_dag_run(dag_run=dag_run, session=session)
ifrun_version_idandrun_version_id!=trigger.task_instance.dag_version_id:
serialized_dag_model= (
dag_bag.get_serialized_dag_model(version_id=run_version_id, session=session)
orserialized_dag_model
)

[seanghaeli]

@hussein-awala good catch, it can crash in this specific sequence. In fact, the current version of main can fail for a similar reason noted in issue #69841. The latest commit introduces a guard so it falls back to running the trigger without Dag context instead of killing the triggerer

[seanghaeli]

If a Dag run without a bundle_version and one of its tasks point at different Dag versions, the trigger should load the latest version. Currently, it loads the task's possibly outdated version.

[seanghaeli]

@o-nikolas this one is similar in spirit to your PR #69941

[o-nikolas]

Seems like a reasonable change semantically. But someone who knows the trigger code better should review to see if this is what we want (it seems like it should be though to me at least). Maybe @vincbeck?

[viiccwen]

left comments.

[viiccwen]

created_dag_version_id is populated even for Dags with disable_bundle_versioning=True; dag_run.bundle_version is what determines
whether the run is pinned. This unconditional preference for created_dag_version_id therefore makes an unpinned start_from_trigger task load the Dag version from when the run was created, even after the scheduler has advanced the unfinished TI's dag_version_id following a reparse.

This differs from DBDagBag._version_from_dag_run(), which intentionally resolves the latest version when bundle_version is absent. Please use the run's created_dag_version_id only when the run is pinned, retain the TI version for unpinned runs, and cover both cases in the regression test.

[seanghaeli]

I believe this is addressed now, could you take a look?

[o-nikolas]

@viiccwen ☝️

[potiuk]

@seanghaeli — There are 3 unresolved review thread(s) on this PR from @viiccwen. Could you either push a fix or reply in each thread explaining why the feedback doesn't apply? Once you believe the feedback is addressed, mark the thread as resolved so the reviewer isn't re-pinged needlessly. Thanks!

---

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.

[o-nikolas]

@viiccwen do the changes look good to you now?

[vatsrahul1001]

LGTM, I think we are good to merge after code owners review @dstandish@hussein-awala

[jason810496]

Thanks for the fix.

IMO, we should introduce something like get_serialized_dag_model_for_run rather than calling the private method of the DagBag.

# airflow-core/src/airflow/models/dagbag.py, next to get_dag_for_rundefget_serialized_dag_model_for_run(
self, dag_run: DagRun, *, session: Session
) ->SerializedDagModel|None:
"""Return the SerializedDagModel for the version a run executes against."""ifversion_id:=self._version_from_dag_run(dag_run=dag_run, session=session):
returnself.get_serialized_dag_model(version_id=version_id, session=session)
returnNone

airflow/airflow-core/src/airflow/models/dagbag.py

Lines 218 to 221 in b3cc48d

	defget_dag_for_run(self, dag_run: DagRun, session: Session) ->SerializedDAG|None:
	ifversion_id:=self._version_from_dag_run(dag_run=dag_run, session=session):
	returnself._get_dag(version_id=version_id, session=session)
	returnNone

[vatsrahul1001]

Moving to 3.3.2 as this is still pending code owner review and do not want to rush on merging this without that

[vatsrahul1001]

@dstandish@hussein-awala can you review this?

[hussein-awala]

Thanks for the fix, resolving the version from the run is the right direction and it
matches what workers already do. One thing worries me though. This applies to every
trigger, not only the start_from_trigger ones, and that makes the get_task call
able to kill the triggerer.

Here is the scenario. Take a DAG in a bundle without versioning, so its runs have
bundle_version = None.

1. my_dag v1 has a deferrable task wait. A run is created at v1 and wait
   defers. The TI is DEFERRED with dag_version_id = v1 and there is a Trigger
   row for it.
2. Someone edits the file and renames wait to wait_for_data. The dag processor
   creates v2 and it becomes the latest. The run is still RUNNING, its
   bundle_version is still None and created_dag_version_id is still v1.
3. The scheduler does not clean the TI up yet.
   _check_for_removed_or_restored_tasks only marks it REMOVED when
   self.state != DagRunState.RUNNING, so nothing happens until the next
   task_instance_scheduling_decisions pass over that run.
4. In the meantime the triggerer builds the workload for that trigger, which happens
   on a restart, on a rolling deploy, or on HA failover after assign_unassigned.
   _version_from_dag_run sees no bundle_version, returns v2 as the latest, and
   serialized_dag_model.dag.get_task("wait") raises TaskNotFound.
5. Nobody catches it. Not _create_workload, not build_trigger_workloads, not
   run_once. TriggererJobRunner.run() logs it and re-raises, so the process exits
   and every other trigger it was running goes down with it. After the restart
   assign_unassigned gives it the same trigger back and it crashes again, until the
   scheduler gets around to marking the TI REMOVED and clean_unused deletes the
   trigger.

Steps 2 and 4 are really the same event in practice. A rolling deploy of new DAG
code is exactly when new versions show up and triggerers restart.

Before this change the version came from the TI, so the task was always there and
step 4 just worked. We already guard this pattern elsewhere. In
task_instances.py the same resolution is followed by
with contextlib.suppress(TaskNotFound).

The simplest fix I can think of is to keep the flag check on the TI version and only
re-resolve for the case the title is about:

serialized_dag_model=dag_bag.get_serialized_dag_model(
version_id=trigger.task_instance.dag_version_id, session=session
)
ifserialized_dag_model:
task=serialized_dag_model.dag.get_task(trigger.task_instance.task_id)
iftask.start_from_trigger:
dag_run=trigger.task_instance.get_dagrun(session=session)
run_version_id=DBDagBag._version_from_dag_run(dag_run=dag_run, session=session)
ifrun_version_idandrun_version_id!=trigger.task_instance.dag_version_id:
serialized_dag_model= (
dag_bag.get_serialized_dag_model(version_id=run_version_id, session=session)
orserialized_dag_model
)

[seanghaeli]

@hussein-awala good catch, it can crash in this specific sequence. In fact, the current version of main can fail for a similar reason noted in issue #69841. The latest commit introduces a guard so it falls back to running the trigger without Dag context instead of killing the triggerer

[seanghaeli]

If a Dag run without a bundle_version and one of its tasks point at different Dag versions, the trigger should load the latest version. Currently, it loads the task's possibly outdated version.

[seanghaeli]

@o-nikolas this one is similar in spirit to your PR #69941

[o-nikolas]

Seems like a reasonable change semantically. But someone who knows the trigger code better should review to see if this is what we want (it seems like it should be though to me at least). Maybe @vincbeck?

[viiccwen]

left comments.

[viiccwen]

created_dag_version_id is populated even for Dags with disable_bundle_versioning=True; dag_run.bundle_version is what determines
whether the run is pinned. This unconditional preference for created_dag_version_id therefore makes an unpinned start_from_trigger task load the Dag version from when the run was created, even after the scheduler has advanced the unfinished TI's dag_version_id following a reparse.

This differs from DBDagBag._version_from_dag_run(), which intentionally resolves the latest version when bundle_version is absent. Please use the run's created_dag_version_id only when the run is pinned, retain the TI version for unpinned runs, and cover both cases in the regression test.

[seanghaeli]

I believe this is addressed now, could you take a look?

[o-nikolas]

@viiccwen ☝️

[potiuk]

@seanghaeli — There are 3 unresolved review thread(s) on this PR from @viiccwen. Could you either push a fix or reply in each thread explaining why the feedback doesn't apply? Once you believe the feedback is addressed, mark the thread as resolved so the reviewer isn't re-pinged needlessly. Thanks!

---

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.

[o-nikolas]

@viiccwen do the changes look good to you now?

[vatsrahul1001]

LGTM, I think we are good to merge after code owners review @dstandish@hussein-awala

[jason810496]

Thanks for the fix.

IMO, we should introduce something like get_serialized_dag_model_for_run rather than calling the private method of the DagBag.

# airflow-core/src/airflow/models/dagbag.py, next to get_dag_for_rundefget_serialized_dag_model_for_run(
self, dag_run: DagRun, *, session: Session
) ->SerializedDagModel|None:
"""Return the SerializedDagModel for the version a run executes against."""ifversion_id:=self._version_from_dag_run(dag_run=dag_run, session=session):
returnself.get_serialized_dag_model(version_id=version_id, session=session)
returnNone

airflow/airflow-core/src/airflow/models/dagbag.py

Lines 218 to 221 in b3cc48d

	defget_dag_for_run(self, dag_run: DagRun, session: Session) ->SerializedDAG|None:
	ifversion_id:=self._version_from_dag_run(dag_run=dag_run, session=session):
	returnself._get_dag(version_id=version_id, session=session)
	returnNone

[vatsrahul1001]

Moving to 3.3.2 as this is still pending code owner review and do not want to rush on merging this without that

[vatsrahul1001]

@dstandish@hussein-awala can you review this?

[hussein-awala]

Thanks for the fix, resolving the version from the run is the right direction and it
matches what workers already do. One thing worries me though. This applies to every
trigger, not only the start_from_trigger ones, and that makes the get_task call
able to kill the triggerer.

Here is the scenario. Take a DAG in a bundle without versioning, so its runs have
bundle_version = None.

1. my_dag v1 has a deferrable task wait. A run is created at v1 and wait
   defers. The TI is DEFERRED with dag_version_id = v1 and there is a Trigger
   row for it.
2. Someone edits the file and renames wait to wait_for_data. The dag processor
   creates v2 and it becomes the latest. The run is still RUNNING, its
   bundle_version is still None and created_dag_version_id is still v1.
3. The scheduler does not clean the TI up yet.
   _check_for_removed_or_restored_tasks only marks it REMOVED when
   self.state != DagRunState.RUNNING, so nothing happens until the next
   task_instance_scheduling_decisions pass over that run.
4. In the meantime the triggerer builds the workload for that trigger, which happens
   on a restart, on a rolling deploy, or on HA failover after assign_unassigned.
   _version_from_dag_run sees no bundle_version, returns v2 as the latest, and
   serialized_dag_model.dag.get_task("wait") raises TaskNotFound.
5. Nobody catches it. Not _create_workload, not build_trigger_workloads, not
   run_once. TriggererJobRunner.run() logs it and re-raises, so the process exits
   and every other trigger it was running goes down with it. After the restart
   assign_unassigned gives it the same trigger back and it crashes again, until the
   scheduler gets around to marking the TI REMOVED and clean_unused deletes the
   trigger.

Steps 2 and 4 are really the same event in practice. A rolling deploy of new DAG
code is exactly when new versions show up and triggerers restart.

Before this change the version came from the TI, so the task was always there and
step 4 just worked. We already guard this pattern elsewhere. In
task_instances.py the same resolution is followed by
with contextlib.suppress(TaskNotFound).

The simplest fix I can think of is to keep the flag check on the TI version and only
re-resolve for the case the title is about:

serialized_dag_model=dag_bag.get_serialized_dag_model(
version_id=trigger.task_instance.dag_version_id, session=session
)
ifserialized_dag_model:
task=serialized_dag_model.dag.get_task(trigger.task_instance.task_id)
iftask.start_from_trigger:
dag_run=trigger.task_instance.get_dagrun(session=session)
run_version_id=DBDagBag._version_from_dag_run(dag_run=dag_run, session=session)
ifrun_version_idandrun_version_id!=trigger.task_instance.dag_version_id:
serialized_dag_model= (
dag_bag.get_serialized_dag_model(version_id=run_version_id, session=session)
orserialized_dag_model
)

[seanghaeli]

@hussein-awala good catch, it can crash in this specific sequence. In fact, the current version of main can fail for a similar reason noted in issue #69841. The latest commit introduces a guard so it falls back to running the trigger without Dag context instead of killing the triggerer

[seanghaeli]

If a Dag run without a bundle_version and one of its tasks point at different Dag versions, the trigger should load the latest version. Currently, it loads the task's possibly outdated version.

[seanghaeli]

@o-nikolas this one is similar in spirit to your PR #69941

[o-nikolas]

Seems like a reasonable change semantically. But someone who knows the trigger code better should review to see if this is what we want (it seems like it should be though to me at least). Maybe @vincbeck?

[viiccwen]

left comments.

[viiccwen]

created_dag_version_id is populated even for Dags with disable_bundle_versioning=True; dag_run.bundle_version is what determines
whether the run is pinned. This unconditional preference for created_dag_version_id therefore makes an unpinned start_from_trigger task load the Dag version from when the run was created, even after the scheduler has advanced the unfinished TI's dag_version_id following a reparse.

This differs from DBDagBag._version_from_dag_run(), which intentionally resolves the latest version when bundle_version is absent. Please use the run's created_dag_version_id only when the run is pinned, retain the TI version for unpinned runs, and cover both cases in the regression test.

[seanghaeli]

I believe this is addressed now, could you take a look?

[o-nikolas]

@viiccwen ☝️

[potiuk]

@seanghaeli — There are 3 unresolved review thread(s) on this PR from @viiccwen. Could you either push a fix or reply in each thread explaining why the feedback doesn't apply? Once you believe the feedback is addressed, mark the thread as resolved so the reviewer isn't re-pinged needlessly. Thanks!

---

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.

[o-nikolas]

@viiccwen do the changes look good to you now?

[vatsrahul1001]

LGTM, I think we are good to merge after code owners review @dstandish@hussein-awala

[jason810496]

Thanks for the fix.

IMO, we should introduce something like get_serialized_dag_model_for_run rather than calling the private method of the DagBag.

# airflow-core/src/airflow/models/dagbag.py, next to get_dag_for_rundefget_serialized_dag_model_for_run(
self, dag_run: DagRun, *, session: Session
) ->SerializedDagModel|None:
"""Return the SerializedDagModel for the version a run executes against."""ifversion_id:=self._version_from_dag_run(dag_run=dag_run, session=session):
returnself.get_serialized_dag_model(version_id=version_id, session=session)
returnNone

airflow/airflow-core/src/airflow/models/dagbag.py

Lines 218 to 221 in b3cc48d

	defget_dag_for_run(self, dag_run: DagRun, session: Session) ->SerializedDAG|None:
	ifversion_id:=self._version_from_dag_run(dag_run=dag_run, session=session):
	returnself._get_dag(version_id=version_id, session=session)
	returnNone

[vatsrahul1001]

Moving to 3.3.2 as this is still pending code owner review and do not want to rush on merging this without that

[vatsrahul1001]

@dstandish@hussein-awala can you review this?

[hussein-awala]

Thanks for the fix, resolving the version from the run is the right direction and it
matches what workers already do. One thing worries me though. This applies to every
trigger, not only the start_from_trigger ones, and that makes the get_task call
able to kill the triggerer.

Here is the scenario. Take a DAG in a bundle without versioning, so its runs have
bundle_version = None.

1. my_dag v1 has a deferrable task wait. A run is created at v1 and wait
   defers. The TI is DEFERRED with dag_version_id = v1 and there is a Trigger
   row for it.
2. Someone edits the file and renames wait to wait_for_data. The dag processor
   creates v2 and it becomes the latest. The run is still RUNNING, its
   bundle_version is still None and created_dag_version_id is still v1.
3. The scheduler does not clean the TI up yet.
   _check_for_removed_or_restored_tasks only marks it REMOVED when
   self.state != DagRunState.RUNNING, so nothing happens until the next
   task_instance_scheduling_decisions pass over that run.
4. In the meantime the triggerer builds the workload for that trigger, which happens
   on a restart, on a rolling deploy, or on HA failover after assign_unassigned.
   _version_from_dag_run sees no bundle_version, returns v2 as the latest, and
   serialized_dag_model.dag.get_task("wait") raises TaskNotFound.
5. Nobody catches it. Not _create_workload, not build_trigger_workloads, not
   run_once. TriggererJobRunner.run() logs it and re-raises, so the process exits
   and every other trigger it was running goes down with it. After the restart
   assign_unassigned gives it the same trigger back and it crashes again, until the
   scheduler gets around to marking the TI REMOVED and clean_unused deletes the
   trigger.

Steps 2 and 4 are really the same event in practice. A rolling deploy of new DAG
code is exactly when new versions show up and triggerers restart.

Before this change the version came from the TI, so the task was always there and
step 4 just worked. We already guard this pattern elsewhere. In
task_instances.py the same resolution is followed by
with contextlib.suppress(TaskNotFound).

The simplest fix I can think of is to keep the flag check on the TI version and only
re-resolve for the case the title is about:

serialized_dag_model=dag_bag.get_serialized_dag_model(
version_id=trigger.task_instance.dag_version_id, session=session
)
ifserialized_dag_model:
task=serialized_dag_model.dag.get_task(trigger.task_instance.task_id)
iftask.start_from_trigger:
dag_run=trigger.task_instance.get_dagrun(session=session)
run_version_id=DBDagBag._version_from_dag_run(dag_run=dag_run, session=session)
ifrun_version_idandrun_version_id!=trigger.task_instance.dag_version_id:
serialized_dag_model= (
dag_bag.get_serialized_dag_model(version_id=run_version_id, session=session)
orserialized_dag_model
)

[seanghaeli]

@hussein-awala good catch, it can crash in this specific sequence. In fact, the current version of main can fail for a similar reason noted in issue #69841. The latest commit introduces a guard so it falls back to running the trigger without Dag context instead of killing the triggerer

[seanghaeli]

If a Dag run without a bundle_version and one of its tasks point at different Dag versions, the trigger should load the latest version. Currently, it loads the task's possibly outdated version.

[seanghaeli]

@o-nikolas this one is similar in spirit to your PR #69941

[o-nikolas]

Seems like a reasonable change semantically. But someone who knows the trigger code better should review to see if this is what we want (it seems like it should be though to me at least). Maybe @vincbeck?

[viiccwen]

left comments.

[viiccwen]

created_dag_version_id is populated even for Dags with disable_bundle_versioning=True; dag_run.bundle_version is what determines
whether the run is pinned. This unconditional preference for created_dag_version_id therefore makes an unpinned start_from_trigger task load the Dag version from when the run was created, even after the scheduler has advanced the unfinished TI's dag_version_id following a reparse.

This differs from DBDagBag._version_from_dag_run(), which intentionally resolves the latest version when bundle_version is absent. Please use the run's created_dag_version_id only when the run is pinned, retain the TI version for unpinned runs, and cover both cases in the regression test.

[seanghaeli]

I believe this is addressed now, could you take a look?

[o-nikolas]

@viiccwen ☝️

[potiuk]

@seanghaeli — There are 3 unresolved review thread(s) on this PR from @viiccwen. Could you either push a fix or reply in each thread explaining why the feedback doesn't apply? Once you believe the feedback is addressed, mark the thread as resolved so the reviewer isn't re-pinged needlessly. Thanks!

---

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.

[o-nikolas]

@viiccwen do the changes look good to you now?

[vatsrahul1001]

LGTM, I think we are good to merge after code owners review @dstandish@hussein-awala

[jason810496]

Thanks for the fix.

IMO, we should introduce something like get_serialized_dag_model_for_run rather than calling the private method of the DagBag.

# airflow-core/src/airflow/models/dagbag.py, next to get_dag_for_rundefget_serialized_dag_model_for_run(
self, dag_run: DagRun, *, session: Session
) ->SerializedDagModel|None:
"""Return the SerializedDagModel for the version a run executes against."""ifversion_id:=self._version_from_dag_run(dag_run=dag_run, session=session):
returnself.get_serialized_dag_model(version_id=version_id, session=session)
returnNone

airflow/airflow-core/src/airflow/models/dagbag.py

Lines 218 to 221 in b3cc48d

	defget_dag_for_run(self, dag_run: DagRun, session: Session) ->SerializedDAG|None:
	ifversion_id:=self._version_from_dag_run(dag_run=dag_run, session=session):
	returnself._get_dag(version_id=version_id, session=session)
	returnNone

[vatsrahul1001]

Moving to 3.3.2 as this is still pending code owner review and do not want to rush on merging this without that

[vatsrahul1001]

@dstandish@hussein-awala can you review this?

[hussein-awala]

Thanks for the fix, resolving the version from the run is the right direction and it
matches what workers already do. One thing worries me though. This applies to every
trigger, not only the start_from_trigger ones, and that makes the get_task call
able to kill the triggerer.

Here is the scenario. Take a DAG in a bundle without versioning, so its runs have
bundle_version = None.

1. my_dag v1 has a deferrable task wait. A run is created at v1 and wait
   defers. The TI is DEFERRED with dag_version_id = v1 and there is a Trigger
   row for it.
2. Someone edits the file and renames wait to wait_for_data. The dag processor
   creates v2 and it becomes the latest. The run is still RUNNING, its
   bundle_version is still None and created_dag_version_id is still v1.
3. The scheduler does not clean the TI up yet.
   _check_for_removed_or_restored_tasks only marks it REMOVED when
   self.state != DagRunState.RUNNING, so nothing happens until the next
   task_instance_scheduling_decisions pass over that run.
4. In the meantime the triggerer builds the workload for that trigger, which happens
   on a restart, on a rolling deploy, or on HA failover after assign_unassigned.
   _version_from_dag_run sees no bundle_version, returns v2 as the latest, and
   serialized_dag_model.dag.get_task("wait") raises TaskNotFound.
5. Nobody catches it. Not _create_workload, not build_trigger_workloads, not
   run_once. TriggererJobRunner.run() logs it and re-raises, so the process exits
   and every other trigger it was running goes down with it. After the restart
   assign_unassigned gives it the same trigger back and it crashes again, until the
   scheduler gets around to marking the TI REMOVED and clean_unused deletes the
   trigger.

Steps 2 and 4 are really the same event in practice. A rolling deploy of new DAG
code is exactly when new versions show up and triggerers restart.

Before this change the version came from the TI, so the task was always there and
step 4 just worked. We already guard this pattern elsewhere. In
task_instances.py the same resolution is followed by
with contextlib.suppress(TaskNotFound).

The simplest fix I can think of is to keep the flag check on the TI version and only
re-resolve for the case the title is about:

serialized_dag_model=dag_bag.get_serialized_dag_model(
version_id=trigger.task_instance.dag_version_id, session=session
)
ifserialized_dag_model:
task=serialized_dag_model.dag.get_task(trigger.task_instance.task_id)
iftask.start_from_trigger:
dag_run=trigger.task_instance.get_dagrun(session=session)
run_version_id=DBDagBag._version_from_dag_run(dag_run=dag_run, session=session)
ifrun_version_idandrun_version_id!=trigger.task_instance.dag_version_id:
serialized_dag_model= (
dag_bag.get_serialized_dag_model(version_id=run_version_id, session=session)
orserialized_dag_model
)

[seanghaeli]

@hussein-awala good catch, it can crash in this specific sequence. In fact, the current version of main can fail for a similar reason noted in issue #69841. The latest commit introduces a guard so it falls back to running the trigger without Dag context instead of killing the triggerer

[seanghaeli]

If a Dag run without a bundle_version and one of its tasks point at different Dag versions, the trigger should load the latest version. Currently, it loads the task's possibly outdated version.

[seanghaeli]

@o-nikolas this one is similar in spirit to your PR #69941

[o-nikolas]

Seems like a reasonable change semantically. But someone who knows the trigger code better should review to see if this is what we want (it seems like it should be though to me at least). Maybe @vincbeck?

[viiccwen]

left comments.

[viiccwen]

created_dag_version_id is populated even for Dags with disable_bundle_versioning=True; dag_run.bundle_version is what determines
whether the run is pinned. This unconditional preference for created_dag_version_id therefore makes an unpinned start_from_trigger task load the Dag version from when the run was created, even after the scheduler has advanced the unfinished TI's dag_version_id following a reparse.

This differs from DBDagBag._version_from_dag_run(), which intentionally resolves the latest version when bundle_version is absent. Please use the run's created_dag_version_id only when the run is pinned, retain the TI version for unpinned runs, and cover both cases in the regression test.

[seanghaeli]

I believe this is addressed now, could you take a look?

[o-nikolas]

@viiccwen ☝️

[potiuk]

@seanghaeli — There are 3 unresolved review thread(s) on this PR from @viiccwen. Could you either push a fix or reply in each thread explaining why the feedback doesn't apply? Once you believe the feedback is addressed, mark the thread as resolved so the reviewer isn't re-pinged needlessly. Thanks!

---

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.

[o-nikolas]

@viiccwen do the changes look good to you now?

[vatsrahul1001]

LGTM, I think we are good to merge after code owners review @dstandish@hussein-awala

[jason810496]

Thanks for the fix.

IMO, we should introduce something like get_serialized_dag_model_for_run rather than calling the private method of the DagBag.

# airflow-core/src/airflow/models/dagbag.py, next to get_dag_for_rundefget_serialized_dag_model_for_run(
self, dag_run: DagRun, *, session: Session
) ->SerializedDagModel|None:
"""Return the SerializedDagModel for the version a run executes against."""ifversion_id:=self._version_from_dag_run(dag_run=dag_run, session=session):
returnself.get_serialized_dag_model(version_id=version_id, session=session)
returnNone

airflow/airflow-core/src/airflow/models/dagbag.py

Lines 218 to 221 in b3cc48d

	defget_dag_for_run(self, dag_run: DagRun, session: Session) ->SerializedDAG|None:
	ifversion_id:=self._version_from_dag_run(dag_run=dag_run, session=session):
	returnself._get_dag(version_id=version_id, session=session)
	returnNone

[vatsrahul1001]

Moving to 3.3.2 as this is still pending code owner review and do not want to rush on merging this without that

[vatsrahul1001]

@dstandish@hussein-awala can you review this?

[hussein-awala]

Thanks for the fix, resolving the version from the run is the right direction and it
matches what workers already do. One thing worries me though. This applies to every
trigger, not only the start_from_trigger ones, and that makes the get_task call
able to kill the triggerer.

Here is the scenario. Take a DAG in a bundle without versioning, so its runs have
bundle_version = None.

1. my_dag v1 has a deferrable task wait. A run is created at v1 and wait
   defers. The TI is DEFERRED with dag_version_id = v1 and there is a Trigger
   row for it.
2. Someone edits the file and renames wait to wait_for_data. The dag processor
   creates v2 and it becomes the latest. The run is still RUNNING, its
   bundle_version is still None and created_dag_version_id is still v1.
3. The scheduler does not clean the TI up yet.
   _check_for_removed_or_restored_tasks only marks it REMOVED when
   self.state != DagRunState.RUNNING, so nothing happens until the next
   task_instance_scheduling_decisions pass over that run.
4. In the meantime the triggerer builds the workload for that trigger, which happens
   on a restart, on a rolling deploy, or on HA failover after assign_unassigned.
   _version_from_dag_run sees no bundle_version, returns v2 as the latest, and
   serialized_dag_model.dag.get_task("wait") raises TaskNotFound.
5. Nobody catches it. Not _create_workload, not build_trigger_workloads, not
   run_once. TriggererJobRunner.run() logs it and re-raises, so the process exits
   and every other trigger it was running goes down with it. After the restart
   assign_unassigned gives it the same trigger back and it crashes again, until the
   scheduler gets around to marking the TI REMOVED and clean_unused deletes the
   trigger.

Steps 2 and 4 are really the same event in practice. A rolling deploy of new DAG
code is exactly when new versions show up and triggerers restart.

Before this change the version came from the TI, so the task was always there and
step 4 just worked. We already guard this pattern elsewhere. In
task_instances.py the same resolution is followed by
with contextlib.suppress(TaskNotFound).

The simplest fix I can think of is to keep the flag check on the TI version and only
re-resolve for the case the title is about:

serialized_dag_model=dag_bag.get_serialized_dag_model(
version_id=trigger.task_instance.dag_version_id, session=session
)
ifserialized_dag_model:
task=serialized_dag_model.dag.get_task(trigger.task_instance.task_id)
iftask.start_from_trigger:
dag_run=trigger.task_instance.get_dagrun(session=session)
run_version_id=DBDagBag._version_from_dag_run(dag_run=dag_run, session=session)
ifrun_version_idandrun_version_id!=trigger.task_instance.dag_version_id:
serialized_dag_model= (
dag_bag.get_serialized_dag_model(version_id=run_version_id, session=session)
orserialized_dag_model
)

[seanghaeli]

@hussein-awala good catch, it can crash in this specific sequence. In fact, the current version of main can fail for a similar reason noted in issue #69841. The latest commit introduces a guard so it falls back to running the trigger without Dag context instead of killing the triggerer