ephraimbuddy commented on code in PR #71696:
URL: https://github.com/apache/airflow/pull/71696#discussion_r3796200845
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -461,10 +492,14 @@ def clear_task_instances(
_recalculate_dagrun_queued_at_deadlines(dr, dr.queued_at, session)
+ # A run with no version of its own has nothing to preserve, so the
latest is all
+ # it can be re-run on. Runs migrated from Airflow 2 are like this,
as are runs
+ # whose version `airflow db clean` has since deleted.
+ use_latest_version = run_on_latest_version or
dr.created_dag_version_id is None
Review Comment:
The `airflow db clean` half of this fix is unreachable through the API, so
the
repair below never runs for it. Cleanup deletes a version referenced only by
a run
(`skip_if_referenced` covers `task_instance.dag_version_id` only, and
`dag_run.created_dag_version_id` is `ON DELETE SET NULL` —
`db_cleanup.py:224-228`), which leaves
`created_dag_version_id` NULL while `bundle_version` keeps its value on a
versioned bundle. For
that combination `_version_from_dag_run` (`models/dagbag.py:211`) skips the
latest-version
fallback because `bundle_version` is truthy and returns NULL, so
`get_dag_for_run` yields nothing
and both clear entry points reject the request with 404 before reaching this
code
(`services/public/dag_run.py:91-93`, `routes/public/task_instances.py:893`).
The migrated-from-Airflow-2 case works only because `bundle_version` is
also NULL there, which is
what the manual verification exercised. Could the fallback in
`_version_from_dag_run` also apply
when `created_dag_version_id` is NULL — `get_dag_for_run_or_latest_version`
already does exactly
that (`api_fastapi/common/dagbag.py:106-111`) — with a route-level
regression for
`bundle_version` set + `created_dag_version_id` NULL?
---
Drafted-by: Claude Code (Opus 5); reviewed by @ephraimbuddy before posting
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -478,14 +513,14 @@ def clear_task_instances(
dr_dag = scheduler_dagbag.get_dag_for_run(dag_run=dr,
session=session)
if not dr_dag:
log.warning("No serialized dag found for dag '%s'",
dr.dag_id)
- if dr_dag and not dr_dag.disable_bundle_versioning and
run_on_latest_version:
+ if dr_dag and not dr_dag.disable_bundle_versioning and
use_latest_version:
Review Comment:
When the selected Dag disables bundle versioning, this leaves any existing
`dr.bundle_version` untouched. That can happen after Dag-version cleanup, which
nulls `created_dag_version_id` without necessarily clearing the bundle version,
and would leave the repaired run pinned to a stale bundle. We can explicitly
assign `None` when bundle versioning is disabled, in both this branch and the
active-run branch below.
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -396,7 +420,14 @@ def clear_task_instances(
# the task is terminated and becomes eligible for retry.
else:
dr = ti.dag_run
- if run_on_latest_version:
+ # Nothing to re-run on but the latest, either because the task
instance has no
+ # version or because its run has none and the run loop below moves
it there. The
+ # two columns disagree more often than you would think: the
scheduler backfills a
+ # version onto a migrated run's task instances but never onto the
run itself.
+ use_latest_version = (
+ run_on_latest_version or ti.dag_version_id is None or
dr.created_dag_version_id is None
Review Comment:
`ti.dag_version_id is None` doesn't imply latest — the run may be pinned.
This condition is wider than the run-level one on line 498, so with
`run_on_latest_version=False` a versionless TI on a run pinned to V1 is moved
to the latest while the run loop leaves the run on V1.
It's reachable from this PR's own scenario: a first clear with
`run_on_latest_version=True` pins the run but deliberately leaves uncleared TIs
versionless (line 511, "Only cleared TIs get latest dag_version_id above"), so
a later plain clear of one of those splits the two. That's the mirror of what
`test_clear_task_instances_keeps_run_and_task_versions_together` asserts, and
it isn't covered. Inheriting `dr.created_dag_version_id` when the run has one
unsticks the TI and keeps them together; only a genuinely versionless run
should force latest.
---
Drafted-by: Claude Code (Opus 5); reviewed by @ephraimbuddy before posting
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
