This is an automated email from the ASF dual-hosted git repository. vatsrahul1001 pushed a commit to branch fix-retry-policy-history-audit in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 9194b52fc4d997126a13d93b0a20fe9215d17ac2 Author: Rahul Vats <[email protected]> AuthorDate: Thu Jul 2 13:49:30 2026 +0530 Fix retry policy overrides not persisted to task instance history 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 | 37 ++++++++++++++++++++++ 2 files changed, 44 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..77fbdd0dcfd 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,43 @@ 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.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(
