Dev-iL commented on code in PR #75396:
URL: https://github.com/apache/airflow/pull/75396#discussion_r4503442564
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py:
##########
@@ -563,31 +581,34 @@ def head_xcom(
]
),
description="Get a single XCom Value",
+ response_model=XComResponse,
)
-def get_xcom(
+async def get_xcom(
dag_id: str,
run_id: str,
task_id: str,
key: Annotated[str, Path(min_length=1)],
- session: SessionDep,
+ session: AsyncSessionDep,
params: Annotated[GetXcomFilterParams, Query()],
dag_bag: DagBagDep,
token: ExecutionToken = _TriggerAwareToken,
-) -> XComResponse:
+) -> JSONResponse:
"""Get an Airflow XCom from database - not other XCom Backends."""
owner = None
if token.claims.scope == "execution":
- owner = session.execute(
- select(TaskInstance.id, TaskInstance.working_set)
- .where(
- TaskInstance.id == token.id,
- TaskInstance.dag_id == dag_id,
- TaskInstance.run_id == run_id,
- TaskInstance.task_id == task_id,
- *build_coordinate_filters(TaskInstance,
map_index=params.map_index),
+ owner = (
+ await session.execute(
+ select(TaskInstance.id, TaskInstance.working_set)
+ .where(
+ TaskInstance.id == token.id,
+ TaskInstance.dag_id == dag_id,
+ TaskInstance.run_id == run_id,
+ TaskInstance.task_id == task_id,
+ *build_coordinate_filters(TaskInstance,
map_index=params.map_index),
+ )
+ .limit(1)
+ .execution_options(include_all_attempts=True)
)
- .limit(1)
- .execution_options(include_all_attempts=True)
).first()
Review Comment:
Good catch. I measured three options before choosing, in Breeze with pool
checkout and check-in listeners on both engines (11 requests, first dropped,
two runs each). Cells are the median milliseconds per request during which an
async and a sync connection were checked out together, run 1; run 2.
| Request | Owner check on the route session (before) | `rollback()` after
it | Owner check in the lookup's session |
| --- | --- | --- | --- |
| Warm read, PostgreSQL | 11.3; 5.4 | 0; 0 | 0; 0 |
| Dag cache cleared, PostgreSQL | 155; 151 | 0; 0 | 0; 0 |
| Dag cache cleared, SQLite | 76; 78 | 0; 0 | 0; 0 |
Event-loop stall didn't differ measurably between the three. The rollback
took a second async checkout per request, with the pre-ping on PostgreSQL, and
cold reads on SQLite got slower (153–172 ms request median against 97–100 ms).
So the check now runs first inside `_build_xcom_read`'s worker-thread session.
It's the same query in the same place in the error order, and adds no checkout.
Measuring this turned up a second case. With `[core] multi_team` on, Cadwyn
solves `has_xcom_access` in both of its dependency passes, and the first pass's
session kept an async connection checked out until the response. A multi-team
read held 2 async connections and 1 sync connection at its peak.
`has_xcom_access` now queries in a short session of its own: peak 1, no
measurable change in stall or latency. A test asserts that no async connection
is checked out during the lookup, with and without multi-team. It fails before
each change.
The trigger-header check still queries the route's session before the
lookup. My harness can't drive the triggerer's in-process API, so I left that
path as it is; it's listed under debt in the description.
---
Drafted-by: Claude Code (Opus 5.5) (no human review before posting)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
