fixup! Read current task executions by exact region and try

The task instance clear endpoint now takes its run locks through the shared
helper, so the clears it calls do not lock the same runs again. It keeps
reloading the locked runs, since the run state it acts on may have changed
while it waited for the lock. The conversion lives here rather than in the
clear layer because this layer rewrote the endpoint's imports.

Reading a loop task's execution without its pass must answer the same 400 as
marking it, whether or not a region id is given; until the read routes took
region coordinates there was no read to pin this on.

Author: ashb

airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py

@@ -24,7 +24,7 @@
 
 import structlog
 from fastapi import Depends, HTTPException, Query, status
-from sqlalchemy import select, tuple_
+from sqlalchemy import select
 from sqlalchemy.orm import joinedload
 
 from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity
@@ -129,6 +129,7 @@
 from airflow.api_fastapi.logging.decorators import action_logging
 from airflow.exceptions import AirflowClearRunningTaskException, TaskNotFound
 from airflow.models import DagRun
+from airflow.models.dagrun import lock_dag_runs
 from airflow.models.dynamic_region import SENTINEL_REGION_ID
 from airflow.models.renderedtifields import load_legacy_rendered_fields
 from airflow.models.task_coordinates import (
@@ -1148,13 +1149,7 @@ def select_candidates() -> Sequence[TI]:
     if task_instances:
         if not dry_run:
             run_keys = {(ti.dag_id, ti.run_id) for ti in task_instances}
-            session.scalars(
-                select(DagRun)
-                .where(tuple_(DagRun.dag_id, DagRun.run_id).in_(run_keys))
-                .order_by(DagRun.dag_id, DagRun.run_id)
-                .with_for_update()
-                .execution_options(populate_existing=True)
-            ).all()
+            lock_dag_runs(session, run_keys, refresh=True)
             if body.task_instance_ids is None:
                 task_instances = [ti for ti in select_candidates() if (ti.dag_id, ti.run_id) in run_keys]
         task_instances = session.scalars(

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py

@@ -5795,6 +5795,20 @@ def test_loop_mark_without_coordinates_requires_a_region_and_index(self, test_cl
         assert response.status_code == 400, response.text
         assert response.json()["detail"] == "Select a region and index for this task instance"
 
+    @pytest.mark.parametrize("with_region_id", [False, True])
+    def test_loop_read_without_index_requires_a_region_and_index(
+        self, test_client, loop_instances, with_region_id
+    ):
+        dr, loop, root, tis = loop_instances
+
+        response = test_client.get(
+            f"/dags/{dr.dag_id}/dagRuns/{dr.run_id}/taskInstances/body.first",
+            params={"region_id": str(root.id)} if with_region_id else {},
+        )
+
+        assert response.status_code == 400, response.text
+        assert response.json()["detail"] == "Select a region and index for this task instance"
+
     @pytest.mark.parametrize(
         ("map_index", "expected_status"),
         [