This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 3bfa5ff9a37 Fix retry policy overrides not persisted to task instance
history (#69235)
3bfa5ff9a37 is described below
commit 3bfa5ff9a373aae6669f7cc7a6782da71e714186
Author: Rahul Vats <[email protected]>
AuthorDate: Thu Jul 2 15:15:13 2026 +0530
Fix retry policy overrides not persisted to task instance history (#69235)
AIP-105's per-try audit trail (retry_delay_override, retry_reason) is meant
to
live durably in task_instance_history — the columns on the live
task_instance
row are transient and cleared when the task next enters RUNNING. The
Execution
API retry handler wrote the overrides only into the live-row UPDATE and
archived
the task instance to history before those values were applied, so the
history
columns were always NULL. That silently dropped the audit trail the feature
advertises, while retry timing (which reads the live row) stayed correct.
Setting the overrides on the task instance before prepare_db_for_next_try()
lets
record_ti() snapshot them into task_instance_history, restoring the per-try
record.
---
.../execution_api/routes/task_instances.py | 7 ++++
.../versions/head/test_task_instances.py | 48 ++++++++++++++++++++++
2 files changed, 55 insertions(+)
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
index 0f862e16258..3b142a4e6b1 100644
---
a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
+++
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
@@ -644,6 +644,13 @@ def _create_ti_state_update_query_and_update_state(
_handle_fail_fast_for_dag(ti=ti, dag_id=dag_id,
session=session, dag_bag=dag_bag)
elif isinstance(ti_patch_payload, TIRetryStatePayload):
if ti is not None:
+ # Set the overrides on the TI *before* archiving so record_ti()
+ # snapshots them into task_instance_history (it copies attrs
off
+ # the ti object). Otherwise the per-try audit trail is always
NULL.
+ ti.retry_delay_override = ti_patch_payload.retry_delay_seconds
+ ti.retry_reason = (
+ ti_patch_payload.retry_reason[:500] if
ti_patch_payload.retry_reason else None
+ )
ti.prepare_db_for_next_try(session=session)
# Store retry policy overrides so next_retry_datetime() can read
them.
# These are cleared when the task enters RUNNING (ti_run).
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
index 67e0a4ee4ae..4920fc001f9 100644
---
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
@@ -1888,6 +1888,44 @@ class TestTIUpdateState:
assert ti.retry_delay_override == 42.5
assert ti.retry_reason == "Rate limit: backing off"
+ def test_ti_update_state_retry_policy_overrides_persisted_in_history(
+ self, client, session, create_task_instance
+ ):
+ """Retry policy override + reason must be archived to
task_instance_history.
+
+ record_ti() snapshots columns off the TI object, so the overrides must
be set on
+ the TI before prepare_db_for_next_try() archives it. When they were
written only
+ to the live-row UPDATE, the per-try audit trail in
task_instance_history was
+ always NULL even though the live row was correct.
+ """
+ ti = create_task_instance(
+ task_id="test_retry_policy_override_history",
+ state=State.RUNNING,
+ )
+ session.commit()
+
+ response = client.patch(
+ f"/execution/task-instances/{ti.id}/state",
+ json={
+ "state": State.UP_FOR_RETRY,
+ "end_date": DEFAULT_END_DATE.isoformat(),
+ "retry_delay_seconds": 42.5,
+ "retry_reason": "Rate limit: backing off",
+ },
+ )
+
+ assert response.status_code == 204
+
+ tih = session.scalars(
+ select(TaskInstanceHistory).where(
+ TaskInstanceHistory.dag_id == ti.dag_id,
+ TaskInstanceHistory.task_id == ti.task_id,
+ TaskInstanceHistory.run_id == ti.run_id,
+ )
+ ).one()
+ assert tih.retry_delay_override == 42.5
+ assert tih.retry_reason == "Rate limit: backing off"
+
def test_ti_update_state_retry_without_policy_overrides(self, client,
session, create_task_instance):
"""Without retry policy fields, the columns remain NULL."""
ti = create_task_instance(
@@ -1913,6 +1951,16 @@ class TestTIUpdateState:
assert ti.retry_delay_override is None
assert ti.retry_reason is None
+ tih = session.scalars(
+ select(TaskInstanceHistory).where(
+ TaskInstanceHistory.dag_id == ti.dag_id,
+ TaskInstanceHistory.task_id == ti.task_id,
+ TaskInstanceHistory.run_id == ti.run_id,
+ )
+ ).one()
+ assert tih.retry_delay_override is None
+ assert tih.retry_reason is None
+
def test_ti_run_clears_retry_policy_overrides(self, client, session,
create_task_instance):
"""When a task enters RUNNING, retry policy overrides from the
previous attempt are cleared."""
ti = create_task_instance(
