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 cff39a568502fc9da0d323e262f9772888469fa1 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 | 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(
