Fix TypeError in TaskInstance.next_retry_datetime() for None end_date

[AdityaBhattacharya1]

A task instance can end up in the up_for_retry state with end_date left as None, for example if the process running it is killed abruptly before the end date gets recorded, or via any code path that transitions task state without going through TaskInstance.set_state(). When that happens, next_retry_datetime() unconditionally computes self.end_date + delay, raising:

TypeError: unsupported operand type(s) for +: 'NoneType' and 'datetime.timedelta'

This is called from NotInRetryPeriodDep._get_dep_statuses(), which the scheduler invokes on every scheduling pass for a retry-eligible task. The exception is unhandled at that call site, so depending on version this either crashes the scheduler process outright, or (after the per-DagRun failure isolation added in #62893) gets caught and logged but causes that DagRun's scheduling pass to fail identically every time it's re-evaluated — the affected task instance never recovers on its own without directly patching its end_date in the metadata database.

Personally I faced this error for a DAG run where a DAG had been transitioned to up_for_retry status and then paused. Later revision of the DAG changed the DAG structurally. When I unpaused the DAG and cleared state for the previous DAG run, the end_date was never repopulated for that task instance: DagRun.verify_integrity() / expand_mapped_task() had to reconcile the stale map_index=-1 row against a task definition that had since become mapped, and that reconciliation left the row up_for_retry with end_date still NULL.

Also added a unit test on next_retry_datetime() covering the end_date=None case directly, and one on NotInRetryPeriodDep reproducing the actual crash path named in the linked issues (state=up_for_retry, end_date=None) to confirm the dependency check no longer raises.

closes: #51640
related: #19615

---

Was generative AI tooling used to co-author this PR?

• Yes

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example Dag that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[yuseok89]

Nice catch on the TypeError.
I might be missing something, but since this is recomputed each scheduler pass, wouldn't an end_date is None TI always land at utcnow() + delay and stay up_for_retry indefinitely? Curious what you think.

[AdityaBhattacharya1]

Solid point, this needs fixing.

I'll change it to set self.end_date = timezone.utcnow() once, instead of a local fallback, so it anchors and the normal backoff math converges the same way it would for a task that just failed. Since this runs inside an active session, the mutation flushes on the caller's next commit without next_retry_datetime() needing a session param of its own.

One trade-off worth flagging explicitly: this persists a synthetic end_date, so duration/Gantt/API will show a "recovery anchor" timestamp rather than a real completion time. That's the same decision #12675 made for the set_state() path. Let me know if that's alright, will implement the changes right away. Thanks!

[yuseok89]

Direction sounds right.
Tiny thing, feel free to ignore. Since is_premature calls next_retry_datetime(), would mutating there mean a plain read touches the DB?
If so, set_state() already anchors this case. Either way works.

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

Line 1020 in 723b06b

self.end_date = self.end_date or current_time

[AdityaBhattacharya1]

Since is_premature calls next_retry_datetime(), would mutating there mean a plain read touches the DB?

Yup, looked into NotInRetryPeriodDep._get_dep_statuses() as an alternative as well but even that is similarly reachable so doesn't really make for a cleaner path. Will go ahead with the aforementioned change 👍🏻

[AdityaBhattacharya1]

Made the change and updated the tests to properly cover this edge case

[yuseok89]

Looks good. Thanks for the fix!

[AdityaBhattacharya1]

Cool! Could you approve the changes then? Can proceed with the merging thereafter

[yuseok89]

I'm a contributor, not a committer, so an "Approve" from me won't unblock the merge. It needs a committer to review and merge.

[AdityaBhattacharya1]

Ahh that makes sense, thanks for letting me know!