This is an automated email from the ASF dual-hosted git repository.
guan404ming 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 f306f1848f6 Refresh HITL review details on task clear (#73235)
f306f1848f6 is described below
commit f306f1848f6a847abe482bee29c52c83f6dfb742
Author: Guan-Ming Chiu <[email protected]>
AuthorDate: Wed Sep 30 23:22:54 2026 +0800
Refresh HITL review details on task clear (#73235)
* Refresh HITL review details on task clear
A cleared or retried HITL task re-runs execute and posts a fresh request,
but the existing row kept the first attempt's subject, body, defaults and
params. Reviewers then saw and approved stale content, and pre-filled params
returned the first attempt's output as the task result.
* Re-read the HITL row under lock before validating a response
* Pin refreshed HITL columns and lock TI explicitly
---
.../api_fastapi/core_api/routes/public/hitl.py | 17 +++----
.../api_fastapi/execution_api/routes/hitl.py | 54 +++++++++++++---------
.../core_api/routes/public/test_hitl.py | 28 +++++++++++
.../execution_api/versions/head/test_hitl.py | 36 +++++++++++++--
4 files changed, 101 insertions(+), 34 deletions(-)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py
index 25862ab527b..4771d1de934 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py
@@ -171,20 +171,21 @@ def update_hitl_detail(
# Execution API park transition, so a human response racing the worker's
park cannot deadlock.
# Locking the TI also serializes respond-vs-clear (the clear path locks
the TI, not the HITL row).
locked_ti = (
- session.get(TI, task_instance.id, with_for_update={"of": TI})
+ session.get(TI, task_instance.id, with_for_update={"of": TI},
populate_existing=True)
if isinstance(task_instance, TI)
else None
)
# Lock the hitl_detail row (FOR UPDATE OF hitl_detail). of= scopes the
lock to hitl_detail, which
# eager-joins task_instance (lazy="joined"); a bare with_for_update()
would emit FOR UPDATE against
- # the nullable side of that outer join, which Postgres rejects. The
joinedloaded relationship object
- # reused below is the same identity-mapped row, now locked for this
transaction.
- session.execute(
+ # the nullable side of that outer join, which Postgres rejects.
populate_existing re-reads the
+ # joinedloaded row under the lock, so assignees and options are validated
against the request committed
+ # by a concurrent upsert from the re-run, not the snapshot taken before
locking.
+ hitl_detail_model = session.scalars(
select(HITLDetailModel)
.where(HITLDetailModel.ti_id == task_instance.id)
.with_for_update(of=HITLDetailModel)
- )
- hitl_detail_model = task_instance.hitl_detail
+ .execution_options(populate_existing=True)
+ ).one()
if hitl_detail_model.response_received:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
@@ -225,10 +226,10 @@ def update_hitl_detail(
"Multiple options chosen but this Human-in-the-loop task accepts
only a single option.",
)
- hitl_detail_model.responded_by = hitl_user
+ hitl_detail_model.responded_by = dict(hitl_user)
hitl_detail_model.responded_at = timezone.utcnow()
hitl_detail_model.chosen_options =
update_hitl_detail_payload.chosen_options
- hitl_detail_model.params_input = update_hitl_detail_payload.params_input
+ hitl_detail_model.params_input =
dict(update_hitl_detail_payload.params_input)
session.add(hitl_detail_model)
# Event-driven resume: if the task is parked waiting for this input,
transition it directly,
diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/routes/hitl.py
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/hitl.py
index 74cda44efe7..5d34cc6921f 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/routes/hitl.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/routes/hitl.py
@@ -32,6 +32,7 @@ from airflow.api_fastapi.execution_api.datamodels.hitl import
(
)
from airflow.api_fastapi.execution_api.security import ExecutionAPIRoute,
require_auth
from airflow.models.hitl import HITLDetail
+from airflow.models.taskinstance import TaskInstance as TI
router = VersionedAPIRouter(
route_class=ExecutionAPIRoute,
@@ -61,34 +62,43 @@ def upsert_hitl_detail(
1. If a HITLOperator task instance does not have a HITLDetail,
a new HITLDetail is created without a response section.
2. If a HITLOperator task instance has a HITLDetail but lacks a response,
- the existing HITLDetail is returned.
+ the request part is refreshed from the payload and the HITLDetail is
returned.
This situation occurs when a task instance is cleared before a response
is received.
3. If a HITLOperator task instance has both a HITLDetail and a response
section,
- the existing response is removed, and the HITLDetail is returned.
+ the request part is refreshed, the existing response is removed, and
the HITLDetail is returned.
This happens when a task instance is cleared after a response has been
received.
This design ensures that each task instance has only one HITLDetail.
"""
- hitl_detail_model =
session.scalar(select(HITLDetail).where(HITLDetail.ti_id == task_instance_id))
+ request_part = {
+ "options": payload.options,
+ "subject": payload.subject,
+ "body": payload.body,
+ "defaults": payload.defaults,
+ "multiple": payload.multiple,
+ "params": payload.params,
+ "assignees": [user.model_dump() for user in payload.assigned_users],
+ "created_at": timezone.utcnow(),
+ }
+ # Same lock order as the park transition and the Core API response path
(TaskInstance, then
+ # hitl_detail), so a response landing between the read and the flush
cannot survive the rewrite.
+ session.get(TI, task_instance_id, with_for_update={"of": TI})
+ hitl_detail_model = session.scalar(
+ select(HITLDetail).where(HITLDetail.ti_id ==
task_instance_id).with_for_update(of=HITLDetail)
+ )
if not hitl_detail_model:
- hitl_detail_model = HITLDetail(
- ti_id=task_instance_id,
- options=payload.options,
- subject=payload.subject,
- body=payload.body,
- defaults=payload.defaults,
- multiple=payload.multiple,
- params=payload.params,
- assignees=[user.model_dump() for user in payload.assigned_users],
- )
- session.add(hitl_detail_model)
- elif hitl_detail_model.response_received:
- # Cleanup the response part of HITLDetail as we only store one
response for one task instance.
- # It normally happens after retry, we keep only the latest response.
- hitl_detail_model.responded_by = None
- hitl_detail_model.responded_at = None
- hitl_detail_model.chosen_options = None
- hitl_detail_model.params_input = {}
- session.add(hitl_detail_model)
+ hitl_detail_model = HITLDetail(ti_id=task_instance_id, **request_part)
+ else:
+ if hitl_detail_model.response_received:
+ # Cleanup the response part of HITLDetail as we only store one
response for one task instance.
+ # It normally happens after retry, we keep only the latest
response.
+ hitl_detail_model.responded_by = None
+ hitl_detail_model.responded_at = None
+ hitl_detail_model.chosen_options = None
+ hitl_detail_model.params_input = {}
+ # A retry re-runs the operator with regenerated content, so the row
must follow the latest request.
+ for column, value in request_part.items():
+ setattr(hitl_detail_model, column, value)
+ session.add(hitl_detail_model)
return HITLDetailRequest.model_validate(hitl_detail_model)
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py
index 75f7d65ecbd..87cd5cbf871 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py
@@ -30,6 +30,7 @@ from sqlalchemy.orm import Session
from airflow._shared.serialization import CLASSNAME, FORBIDDEN_XCOM_KEYS
from airflow._shared.timezones.timezone import utc, utcnow
+from airflow.api_fastapi.core_api.routes.public import hitl as hitl_routes
from airflow.models.dag import DagModel
from airflow.models.dagbundle import DagBundleModel
from airflow.models.hitl import HITLDetail
@@ -426,6 +427,33 @@ class TestUpdateHITLDetailEndpoint:
assert response.status_code == 400
assert "Invalid options" in response.json()["detail"]
+ @pytest.mark.usefixtures("sample_hitl_detail")
+ def test_should_validate_against_row_refreshed_under_lock(
+ self,
+ test_client: TestClient,
+ sample_ti_url_identifier: str,
+ sample_ti: TaskInstance,
+ ) -> None:
+ original = hitl_routes._get_task_instance_with_hitl_detail
+
+ def load_then_rewrite_options(**kwargs: Any) -> Any:
+ ti = original(**kwargs)
+ kwargs["session"].execute(
+ update(HITLDetail).where(HITLDetail.ti_id ==
sample_ti.id).values(options=["Retry"]),
+ execution_options={"synchronize_session": False},
+ )
+ return ti
+
+ with mock.patch.object(
+ hitl_routes, "_get_task_instance_with_hitl_detail",
side_effect=load_then_rewrite_options
+ ):
+ response = test_client.patch(
+ f"{sample_ti_url_identifier}/hitlDetails",
+ json={"chosen_options": ["Retry"], "params_input": {}},
+ )
+ assert response.status_code == 200
+ assert response.json()["chosen_options"] == ["Retry"]
+
@pytest.mark.usefixtures("sample_hitl_detail")
@pytest.mark.parametrize("reserved_key", sorted(FORBIDDEN_XCOM_KEYS))
@pytest.mark.parametrize(
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_hitl.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_hitl.py
index 0b51dfead30..aeda1196886 100644
---
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_hitl.py
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_hitl.py
@@ -22,6 +22,7 @@ from typing import TYPE_CHECKING, Any
import pytest
import time_machine
from httpx import Client
+from sqlalchemy import select
from uuid6 import uuid7
from airflow._shared.timezones.timezone import convert_to_utc
@@ -103,6 +104,7 @@ def expected_sample_hitl_detail_dict(sample_ti:
TaskInstance) -> dict[str, Any]:
"existing hitl detail with response",
],
)
+@time_machine.travel(datetime(2025, 7, 4, 0, 0, 0), tick=False)
def test_upsert_hitl_detail(
client: TestClient,
create_task_instance: CreateTaskInstance,
@@ -113,26 +115,52 @@ def test_upsert_hitl_detail(
session.commit()
if existing_hitl_detail_args:
- session.add(HITLDetail(ti_id=ti.id, **existing_hitl_detail_args))
+ session.add(
+ HITLDetail(
+ ti_id=ti.id,
+ created_at=convert_to_utc(datetime(2025, 7, 1, 0, 0, 0)),
+ **existing_hitl_detail_args,
+ )
+ )
session.commit()
+ request_kwargs = {
+ "subject": "Regenerated subject",
+ "body": "regenerated body",
+ "options": ["Retry", "Skip"],
+ "defaults": ["Skip"],
+ "multiple": True,
+ "params": {"input_1": 3},
+ "assigned_users": [{"id": "2", "name": "reviewer"}],
+ }
response = client.post(
f"/execution/hitlDetails/{ti.id}",
json={
"ti_id": str(ti.id),
- **default_hitl_detail_request_kwargs,
+ **request_kwargs,
},
)
expected_json = {
"ti_id": str(ti.id),
- **default_hitl_detail_request_kwargs,
+ **request_kwargs,
}
- expected_json["assigned_users"] = expected_json.pop("assignees") or []
assert response.status_code == 201
assert response.json() == expected_json
+ session.expire_all()
+ hitl_detail = session.scalars(select(HITLDetail).where(HITLDetail.ti_id ==
ti.id)).one()
+ assert hitl_detail.subject == "Regenerated subject"
+ assert hitl_detail.body == "regenerated body"
+ assert hitl_detail.options == ["Retry", "Skip"]
+ assert hitl_detail.defaults == ["Skip"]
+ assert hitl_detail.multiple is True
+ assert hitl_detail.params == {"input_1": 3}
+ assert hitl_detail.assignees == [{"id": "2", "name": "reviewer"}]
+ assert hitl_detail.created_at == convert_to_utc(datetime(2025, 7, 4, 0, 0,
0))
+ assert hitl_detail.response_received is False
+
def test_upsert_hitl_detail_with_empty_option(
client: TestClient,
