Repository navigation
Conversation
0b8adb7 to
e36fac8
Compare
github-actions
Bot
added
the
fork ci required
|
Fork CI: green. CI for the latest commit passed in your fork. The pull request is ready for maintainer review. Pull requests from contributors run their CI in the contributor's own fork. GitHub Actions is free and has no minute limit on public forks, while the CI capacity of the Apache Software Foundation is shared by every ASF project and is finite. Latest run: https://github.com/Dev-iL/airflow/actions/runs/42412518363 How this works: fork CI for external contributors. This comment was posted automatically by the fork CI monitor from templates drafted with an AI-assisted tool, so it may contain mistakes. An Apache Airflow maintainer, a real person, will review the pull request once its fork CI is green. Why the first pass is automated. |
e36fac8 to
a3fe4cc
Compare
github-actions
Bot
added
the
ready for maintainer review
a3fe4cc to
9679c1a
Compare
github-actions
Bot
added
ready for maintainer review
shahar1
left a comment
shahar1
left a comment
There was a problem hiding this comment.
The conversion and the offloading read correctly, and I found no break in the HTTP contract or the pool-wait reasoning, but XCom writes now go through an async engine that lacks the sync engine's MySQL session setup, which stores wrong timestamps on a non-UTC server. That should be settled before this merges.
XCom writes use an async engine without the MySQL session setup (airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py:790)
Details are in the inline comment. The sync engine's setup carries these two source comments:
this ensures coherence in mysql when storing datetimes (not required for postgres)
(airflow-core/src/airflow/utils/orm_event_handlers.py)
The default isolation level for MySQL (REPEATABLE READ) can introduce inconsistencies when running multiple schedulers, as repeated queries on the same session may read from stale snapshots.
(airflow-core/src/airflow/settings.py)
Smaller observations
xcoms.py:612(inline): the owner check keeps its async connection checked out while the lookup runs on a worker thread.- PR description, benchmark table: the PostgreSQL large-value row reports a median of about 165 ms but a maximum of 553 and 556 ms in the two runs. SQLite's maximum is 156 and 175 ms. The stall is still well below
main, but a near-identical half-second outlier in both runs looks systematic. Do you know what causes it? The harness isn't in the PR, so reviewers can't check.
This review was drafted by an AI-assisted tool and
confirmed by an Apache Airflow maintainer. The findings
below are observations, not blockers; an Apache Airflow
maintainer — a real person — will take the next look at the
PR. If you think a finding is mis-applied, please reply on
the PR and a maintainer will weigh in.More on how Apache Airflow handles maintainer review:
Contributing guide.
| dag_result=dag_result, | ||
| mapped_length=mapped_length, | ||
| session=session, | ||
| await session.run_sync( |
There was a problem hiding this comment.
major: this write now runs on the async engine, which doesn't get the sync engine's MySQL session setup:
configure_orm()callssetup_event_handlers(engine)for the sync engine only, soSET time_zone = '+00:00'never runs on async connections.prepare_engine_args()setsisolation_level="READ COMMITTED"for the sync engine only, so async sessions use the server default.
I ran Airflow's own engines against MySQL 8.0 started with --default-time-zone=+02:00, writing a UtcDateTime column through each:
sync session vars: ('+00:00', 'READ-COMMITTED')
async session vars: ('+02:00', 'REPEATABLE-READ')
async write read by sync session: wrote 17:48:52+00:00 read 15:48:52+00:00
sync write read by sync session: wrote 17:48:52+00:00 read 17:48:52+00:00
This probe covers the engines rather than this route, but set_for_attempt binds timestamp the same way. On such a server, xcom_v2.timestamp for every pushed XCom would be off by the server's UTC offset. That value is shown in the UI and REST API, and airflow db clean uses it as the recency column for xcom_v2. A server on UTC hides this.
The isolation level also matters on the read side: in get_xcom the owner check starts the snapshot, so the value read after the lookup can't see a row committed in between. The window is short, and I only expect a transient 404 from it.
In _configure_async_session, passing isolation_level="READ COMMITTED" for MySQL and calling setup_event_handlers(async_engine.sync_engine) fixed both on that server (('+00:00', 'READ-COMMITTED'), zero delta). I only tried this with asyncmy on MySQL. The handlers would also run for SQLite and PostgreSQL, so those need checking too.
ti_heartbeat already writes through this engine, so a separate PR that this one rebases on seems cleanest.
There was a problem hiding this comment.
Thanks for running it against a non-UTC server. The time-zone half was fixed on main in #76505, which this branch's base already includes. The isolation level was still missing, and I fixed it separately in #76694 as you suggested. There, the async engine gets the same READ COMMITTED setting as the sync one through a shared helper. Its test reads @@session.transaction_isolation through settings.AsyncSession for both aiomysql and asyncmy. That test fails on main with REPEATABLE-READ and passes with the fix. This PR depends on it for MySQL parity, so #76694 should merge first.
Drafted-by: Claude Code (Opus 5.5) reviewed by @Dev-iL before posting.
| ) | ||
| .limit(1) | ||
| .execution_options(include_all_attempts=True) | ||
| ).first() |
There was a problem hiding this comment.
minor: the first row of your connection table comes from this query: its transaction stays open, so the async connection stays checked out and idle in transaction through the run_in_threadpool below, including a cold-cache Dag deserialization. Behind PgBouncer in transaction mode, that also pins a server connection.
owner is a plain row, so await session.rollback() right after this read would release the connection, and this path would no longer hold two at once. The cost is another checkout for the value read, with a pool_pre_ping round trip, so it's your call whether that trade is worth it.
There was a problem hiding this comment.
Good catch. I measured the alternatives before choosing, in Breeze with pool checkout and check-in listeners on both engines: 11 sequential requests per scenario, the first dropped, two runs each (run 1; run 2).
The owner check. Both fixes remove the overlap, so what decides between them is what each costs:
| Owner check | Checkouts per request (async + sync) | Peak connections | Both pools held, PostgreSQL, Dag cache cleared (median ms) | Request time, SQLite, Dag cache cleared (median ms) |
|---|---|---|---|---|
| On the route session (before) | 1 + 1 | 2 | 155; 151 | 99.8; 97.3 |
On the route session, then rollback() |
2 + 1 | 1 | 0; 0 | 172; 153 |
| In a worker-thread session before the lookup | 1 + 2 | 1 | 0; 0 | 92.9; 98.6 |
The rollback releases the connection, but the route then checks out a second one, with the pre-ping on PostgreSQL, and cold reads on SQLite got slower. Moving the check to a worker thread had no such cost. This PR goes one step further than the measured version and runs the check first inside _build_xcom_read's own session, so it keeps the same point in the error order without a second sync checkout: 1 + 1 per request, with no overlap, in the final benchmarks.
The two dependency checks. Measuring this turned up the same pattern twice more. Cadwyn solves has_xcom_access in both of its dependency passes, and with [core] multi_team on, the first pass's session stayed checked out until the response. _resolve_trigger_token, which checks a trigger's claimed identity, queried the route's session before the lookup. Both now query in a short create_session_async() session of their own:
| Request | Peak connections, before → after | Both pools held, before → after (median ms, PostgreSQL; SQLite) | Async checkouts per request, before → after |
|---|---|---|---|
Read with [core] multi_team on |
3 → 1 | 14.3 and 6.3; 9.9 and 12.0 → 0 | 2 → 3 |
Trigger read (GET with Airflow-Trigger-Ti-Id) |
2 → 1 | 15.6 and 14.4; 11.5 and 12.8 → 0 | 1 → 2 |
Trigger write (POST with Airflow-Trigger-Ti-Id) |
2 → 1 | 4.3 and 3.9; 1.1 and 1.5 → 0 | 1 → 2 |
The trigger requests go through InProcessExecutionAPI with the header, as the triggerer sends them. The extra checkouts are short and taken before the lookup. Event-loop stall and request time didn't differ measurably in any of these cases: the medians moved by less than their run-to-run spread. For a trigger's write, the deferred-state check and the write now run in separate transactions. The check never locked anything (it's a plain SELECT), so the task instance's state could already change between them before.
Two tests pin this: test_xcom_read_holds_no_route_connection_during_the_producer_lookup (without and with multi-team) and test_trigger_xcom_request_holds_no_route_connection_during_the_producer_lookup (GET and POST) assert that no async connection is checked out while the lookup runs. Each case fails without its change.
Drafted-by: Claude Code (Opus 5.5) (no human review before posting)
9679c1a to
81d58f1
Compare
github-actions
Bot
removed
the
ready for maintainer review
On the ~550 ms PostgreSQL maximum for the large value: I re-measured and couldn't reproduce it, and I can't name its cause. I re-ran it in Breeze, PostgreSQL, Python 3.14.10, SQLAlchemy 2.0.55, alternating base (
Longer head runs stayed in the same range: 40 warm requests twice, 160 / 234 and 171 / 251, then 150 warm requests, 156 / 259. Across 230 warm requests the largest stall was 259 ms. With the review changes, measured at What I could rule out, from the probe below, which records garbage-collector pauses and which thread they ran on:
On the loop, each large read now spends about 95 ms in psycopg's JSON decoding and about 60 + 57 ms in Cadwyn's decode and re-encode of the Probe (disposable, not part of the PR)Copy it to breeze run --python 3.14 --backend postgres bash -c "python dev/xcom_stall_probe.py --scenario big --requests 11 --label head --out dev/xcom_stall_out.jsonl"#!/usr/bin/env python
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
"""
Event-loop stall probe for the Execution API XCom routes (disposable; not part of any PR).
Run inside Breeze from the checkout root, e.g.:
breeze run --python 3.14 --backend postgres bash -c "python dev/xcom_stall_probe.py \
--scenario big --requests 11 --label head-run1 --out dev/xcom_stall_out.jsonl"
The probe is injected into pytest collection of the Execution API XCom test module, so it gets that
directory's ``client``, ``dag_maker``, ``session`` and ``exec_app`` fixtures. A 1 ms ticker on the
server's event loop records each sleep's own duration; the maximum during a request is its stall. The
HTTP request goes through ``client`` from a worker thread, so routing, dependencies, validation and
serialization are all measured. Every response's status and size are asserted.
Per request it also records:
- connections: checkouts from the sync and async pools, and the peak number held at once;
- async_idle_ms: time the async connection stayed checked out while a sync connection was also out;
- gc: garbage-collector pauses, with the thread they ran on (loop thread or not).
"""
from __future__ import annotations
import argparse
import asyncio
import gc
import json
import os
import statistics
import sys
import threading
import time
import pytest
from fastapi import Request
TEST_MODULE = "airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py"
SCENARIOS = ("big", "small-owner", "small-multiteam", "loop-cold", "loop-warm")
def parse_args():
parser = argparse.ArgumentParser()
parser.add_argument("--scenario", choices=SCENARIOS, required=True)
parser.add_argument("--requests", type=int, default=11)
parser.add_argument("--label", required=True)
parser.add_argument("--out", required=True)
parser.add_argument("--profile", action="store_true", help="profile the loop thread on 1 extra request")
parser.add_argument("--loop-tasks", type=int, default=1000)
return parser.parse_args()
ARGS = parse_args()
class PoolTracker:
"""Track checkouts across the sync and async pools of the metadata DB."""
def __init__(self):
self.lock = threading.Lock()
self.reset()
def reset(self):
with self.lock:
self.held = {"sync": 0, "async": 0}
self.checkouts = {"sync": 0, "async": 0}
self.peak = 0
self.async_out_since = None
self.overlap_start = None
self.overlap_ms = 0.0
def _total(self):
return self.held["sync"] + self.held["async"]
def _update_overlap(self, now):
both = self.held["sync"] > 0 and self.held["async"] > 0
if both and self.overlap_start is None:
self.overlap_start = now
elif not both and self.overlap_start is not None:
self.overlap_ms += (now - self.overlap_start) * 1000
self.overlap_start = None
def install(self, sync_engine, async_engine):
from sqlalchemy import event
for kind, engine in (("sync", sync_engine), ("async", async_engine.sync_engine)):
def on_checkout(dbapi_conn, record, proxy, kind=kind):
with self.lock:
self.held[kind] += 1
self.checkouts[kind] += 1
self.peak = max(self.peak, self._total())
self._update_overlap(time.monotonic())
def on_checkin(dbapi_conn, record, kind=kind):
with self.lock:
self.held[kind] = max(0, self.held[kind] - 1)
self._update_overlap(time.monotonic())
event.listen(engine.pool, "checkout", on_checkout)
event.listen(engine.pool, "checkin", on_checkin)
def snapshot(self):
with self.lock:
if self.overlap_start is not None:
self._update_overlap(time.monotonic())
return {
"checkouts": dict(self.checkouts),
"peak_held": self.peak,
"both_pools_held_ms": round(self.overlap_ms, 2),
}
class GcTracker:
def __init__(self):
self.events = []
self.loop_thread = None
self._start = {}
def __call__(self, phase, info):
tid = threading.get_ident()
if phase == "start":
self._start[tid] = time.monotonic()
elif tid in self._start:
ms = (time.monotonic() - self._start.pop(tid)) * 1000
self.events.append(
{"gen": info["generation"], "ms": round(ms, 2), "on_loop": tid == self.loop_thread}
)
def take(self):
events, self.events = self.events, []
return events
def seed_big(dag_maker, session):
from airflow.models.xcom import XComModel
from airflow.providers.standard.operators.empty import EmptyOperator
with dag_maker(dag_id="stall_big", serialized=True):
EmptyOperator(task_id="producer") >> EmptyOperator(task_id="consumer")
dr = dag_maker.create_dagrun()
tis = {ti.task_id: ti for ti in dr.task_instances}
value = [{"index": i, "name": f"item-{i:06d}"} for i in range(100_000)]
XComModel.set_for_attempt(
task_instance_id=tis["producer"].id, key="big", value=value, serialize=False, session=session
)
session.commit()
url = f"/execution/xcoms/{dr.dag_id}/{dr.run_id}/producer/big"
return url, tis["consumer"], lambda body: len(body["value"]) == 100_000
def seed_small(dag_maker, session):
from airflow.models.xcom import XComModel
from airflow.providers.standard.operators.empty import EmptyOperator
with dag_maker(dag_id="stall_small", serialized=True):
EmptyOperator(task_id="producer") >> EmptyOperator(task_id="consumer")
dr = dag_maker.create_dagrun()
tis = {ti.task_id: ti for ti in dr.task_instances}
XComModel.set_for_attempt(
task_instance_id=tis["producer"].id, key="small", value={"a": 1}, serialize=False, session=session
)
session.commit()
url = f"/execution/xcoms/{dr.dag_id}/{dr.run_id}/producer/small"
return url, tis["consumer"], lambda body: body["value"] == {"a": 1}
def seed_loop(dag_maker, session):
from airflow.models.dynamic_region import DynamicRegion
from airflow.models.xcom import XComModel
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.sdk import task_group
from airflow.sdk.definitions._internal.loop import create_loop
@task_group
def body():
EmptyOperator(task_id="producer") >> EmptyOperator(task_id="consumer")
with dag_maker(dag_id="stall_loop", serialized=True):
start = EmptyOperator(task_id="outside")
start >> create_loop(body, max_iterations=3)
for i in range(ARGS.loop_tasks):
start >> EmptyOperator(task_id=f"filler_{i:04d}")
dr = dag_maker.create_dagrun()
region = DynamicRegion.get_or_create(dag_id=dr.dag_id, run_id=dr.run_id, node_id="body", session=session)
session.add(region)
session.flush()
tis = {ti.task_id: ti for ti in dr.task_instances}
producer, consumer = tis["body.producer"], tis["body.consumer"]
producer.region_id, producer.region_index = region.id, 1
consumer.region_id, consumer.region_index = region.id, 1
session.flush()
XComModel.set_for_attempt(
task_instance_id=producer.id, key="key", value="current", serialize=False, session=session
)
session.commit()
url = f"/execution/xcoms/{dr.dag_id}/{dr.run_id}/{producer.task_id}/key"
return url, consumer, lambda body: body["value"] == "current"
def probe(client, dag_maker, session, exec_app):
from airflow import settings
from airflow.api_fastapi.execution_api.datamodels.token import TIClaims, TIToken
from airflow.api_fastapi.execution_api.security import require_auth
from tests_common.test_utils.config import conf_vars
scenario = ARGS.scenario
seed = {"big": seed_big, "loop-cold": seed_loop, "loop-warm": seed_loop}.get(scenario, seed_small)
url, caller, check = seed(dag_maker, session)
session.close()
async def authenticated(request: Request) -> TIToken:
return TIToken(id=caller.id, claims=TIClaims(scope="execution"))
exec_app.dependency_overrides[require_auth] = authenticated
pools = PoolTracker()
pools.install(settings.engine, settings.async_engine)
gc_tracker = GcTracker()
gc.callbacks.append(gc_tracker)
import sqlalchemy
env = {
"python": sys.version.split()[0],
"sqlalchemy": sqlalchemy.__version__,
"async_driver": settings.async_engine.url.drivername,
"sync_driver": settings.engine.url.drivername,
"loadavg": os.getloadavg(),
}
def one_request(profile=False):
if scenario == "loop-cold":
exec_app.state.dag_bag.clear_cache()
pools.reset()
gc_tracker.take()
async def run():
gc_tracker.loop_thread = threading.get_ident()
stalls: list[float] = []
done = asyncio.Event()
async def ticker():
while not done.is_set():
t0 = time.monotonic()
await asyncio.sleep(0.001)
stalls.append((time.monotonic() - t0) * 1000)
task = asyncio.create_task(ticker())
await asyncio.sleep(0.01)
stalls.clear()
prof = None
if profile:
import cProfile
prof = cProfile.Profile()
prof.enable()
t0 = time.monotonic()
response = await asyncio.to_thread(client.get, url)
elapsed = (time.monotonic() - t0) * 1000
if prof is not None:
prof.disable()
done.set()
await task
return response, elapsed, stalls, prof
response, elapsed, stalls, prof = client.portal.call(run)
assert response.status_code == 200, response.text[:500]
assert check(response.json()), response.text[:200]
return {
"request_ms": round(elapsed, 2),
"stall_max_ms": round(max(stalls), 2) if stalls else 0.0,
"ticks": len(stalls),
"bytes": len(response.content),
**pools.snapshot(),
"gc": gc_tracker.take(),
}, prof
ctx = conf_vars({("core", "multi_team"): "True"}) if scenario == "small-multiteam" else None
if ctx is not None:
ctx.__enter__()
try:
samples = [one_request()[0] for _ in range(ARGS.requests)]
profile_text = None
if ARGS.profile:
import io
import pstats
_, prof = one_request(profile=True)
buf = io.StringIO()
pstats.Stats(prof, stream=buf).sort_stats("tottime").print_stats(25)
profile_text = buf.getvalue()
finally:
if ctx is not None:
ctx.__exit__(None, None, None)
gc.callbacks.remove(gc_tracker)
warm = samples[1:]
summary = {
"label": ARGS.label,
"scenario": scenario,
"env": env,
"stall_median_ms": round(statistics.median(s["stall_max_ms"] for s in warm), 2),
"stall_max_ms": round(max(s["stall_max_ms"] for s in warm), 2),
"request_median_ms": round(statistics.median(s["request_ms"] for s in warm), 2),
"peak_held_max": max(s["peak_held"] for s in warm),
"checkouts_last": warm[-1]["checkouts"],
"both_pools_held_median_ms": round(statistics.median(s["both_pools_held_ms"] for s in warm), 2),
"samples": samples,
"profile": profile_text,
}
with open(ARGS.out, "a") as f:
f.write(json.dumps(summary) + "\n")
print(
f"RESULT {ARGS.label} {scenario}: stall median/max {summary['stall_median_ms']}/"
f"{summary['stall_max_ms']} ms, request median {summary['request_median_ms']} ms, "
f"peak held {summary['peak_held_max']}, both-pools median {summary['both_pools_held_median_ms']} ms, "
f"per-request stalls {[s['stall_max_ms'] for s in samples]}"
)
class _Swap:
@pytest.hookimpl(trylast=True)
def pytest_collection_modifyitems(self, session, config, items):
anchor = next(item for item in items if isinstance(item.parent, pytest.Module))
items[:] = [pytest.Function.from_parent(anchor.parent, name="xcom_stall_probe", callobj=probe)]
if __name__ == "__main__":
sys.exit(pytest.main([TEST_MODULE, "-x", "-q", "-p", "no:cacheprovider", "-s"], plugins=[_Swap()]))Drafted-by: Claude Code (Opus 5.5) (no human review before posting) |
Synchronous XCom routes hold a threadpool worker for every database wait. Awaiting the database lets other Execution API requests make progress while a query waits. On an async route, CPU work runs on the event loop and stalls every request it serves. Producer resolution can deserialize a Dag on a DagBag cache miss, and FastAPI re-validates every element of a large XCom value as a JsonValue, so both stay on worker threads. Checks that run before the producer lookup release their connection first, so a request does not keep an async connection idle while the lookup holds a sync one.
81d58f1 to
4531b90
Compare
github-actions
Bot
added
the
ready for maintainer review
related: #67799, #67800, #73405, #76505, #76694
Convert the six Execution API XCom routes (
GET .../item/{offset},GET .../slice,HEAD,GET,POSTandDELETEon/execution/xcoms/{dag_id}/{run_id}/{task_id}/{key}) toAsyncSessionDep, so database waits release threadpool worker capacity.The HTTP contract is unchanged: paths, parameters, status codes and error bodies, team-scoped access, trigger-identity checks, loop and mapped-producer resolution, slice and offset semantics, and every API version's behaviour. The OpenAPI document is byte-identical to
mainfor all 11 Execution API versions. The session dependency still owns commit and rollback before the response is sent.Important
Each XCom request now uses two database connections instead of one: one from the sync pool and one from the async pool.
Producer resolution runs in a worker thread with its own sync session, so a Dag deserialized on a DagBag cache miss doesn't block the event loop. The route itself uses an
AsyncSession. No request holds both at the same time.[core] multi_teamonGETof a task's own XComDetails & Benchmarks
What runs where
has_xcom_access, the router dependency, queries the two team names in a shortcreate_session_async()session of its own. Cadwyn solves router dependencies in two passes, and the first pass's request-scoped session would otherwise keep an async connection checked out until the response, across the producer lookup._resolve_trigger_token, which checks the task instance a trigger's request claims to act for, also queries in a short session of its own, so the route's session isn't checked out until after the lookup. For a trigger's write, the deferred-state check and the write run in separate transactions; the check is a plainSELECTthat never locked the row, so the state could already change between them before._build_xcom_read,_find_writer_id) runs inrun_in_threadpooland opens its own sync session. It goes throughTaskCoordinateResolver, which can deserialize a serialized Dag on a DagBag cache miss. UnderAsyncSession.run_sync, that CPU work would run on the event loop. This is a compatibility bridge: it uses worker-thread capacity as before. The two helpers return a statement or an id, never ORM objects._build_xcom_read's session, so it keeps the same place in the error order without leaving an async connection idle during the lookup.XComModel.set_for_attemptanddelete_for_attemptsstay single implementations shared with sync callers, called throughrun_syncon the route's session.get_xcomand the item route select only thevaluecolumn instead of hydrating the XCom projection, and they test the fetched row so that a storedNULLvalue still returns 200 rather than 404. The slice reads its values withscalars().JSONResponsebuilt inrun_in_threadpool. The values come from a JSON column, so this skips FastAPI's re-validation of every element asJsonValue, which anasync defroute would otherwise do on the event loop.response_modelstays on each decorator, which keeps the schema. No Cadwyn version change migrates these responses. Cadwyn still decodes and re-encodes a returnedJSONResponsebody on the loop (Optimize request migration for migration-less cases zmievsa/cadwyn#51).Both pools are sized per API server process by
[database] sql_alchemy_pool_size(default 5) plussql_alchemy_max_overflow(default 10). In the worst case a process holds 15 + 15 metadata DB connections, where it used to hold at most 15 sync connections for these routes. Budget that across workers and replicas, and with PgBouncer if you use it. Pool waits cannot deadlock: a request holding an async connection may wait for a sync one, but the sync-only lookup never waits for the async pool. ForPOSTandDELETE, the target task instance is resolved in the lookup's transaction and written in the route's. No row lock is taken on that path, so none is released early.Benchmarks
Measured with a disposable harness that drives the real HTTP stack while a 1 ms ticker on the server loop records each sleep. There were 11 sequential requests per scenario, the first dropped, and each revision ran twice in alternating order. Cells are median / maximum stall in ms for run 1; run 2. Base is
mainat2e9ae2a199; this PR is the same change on that base (b63da48219), before a rebase that brought no XCom, session or settings changes. Python 3.14.10 and SQLAlchemy 2.0.55 in Breeze.GET, one 3.9 MB valueGET, one 3.9 MB valueGETof a task's own small valueGETof a task's own small valueGET,[core] multi_teamonGET,[core] multi_teamonThe large-value request's median time fell from 1443 and 1377 ms to 481 and 512 ms on PostgreSQL, and from 1299 and 1311 ms to 325 and 429 ms on SQLite. The harness also counts connections checked out at once: this PR peaks at 1 in every scenario, while the base peaks at 2 for multi-team reads.
A plain
AsyncSession.run_syncconversion, measured first, regressed the large value (1156–1847 ms median) and the cold-cache loop read (71–110 ms). The loop-thread profile put 1.1 s of the 2.1 s large-value request in pydantic'sJsonValuevalidation. Small reads show medians between about 6 and 10 ms on both revisions. In this session this PR was 1–4 ms higher on PostgreSQL, but differences of that size changed sign between runs and sessions: a same-session rerun of this PR against its revision before the review changes put the warm loop read at 9.2 and 5.7 ms against 8.9 and 10.2 ms. These samples cannot separate the two at that scale. These are small local samples of event-loop availability, not throughput claims. The harness is not part of this PR.An earlier revision of this description showed a 553 ms maximum for the large value on PostgreSQL. Re-measured over 230 warm requests it did not recur: the base pays a 400–730 ms full garbage collection on the loop during every warm large-value request, while this PR's full collections land on the worker thread and did not raise the loop stall.
Tests
has_xcom_accessdirectly, since it opens its own session.test_xcom_read_returns_a_stored_null_without_loading_xcom_entities(getanditem) fails onmain, where each read loads anXComModel.test_xcom_producer_lookups_and_rendering_run_off_the_event_loop(six routes) asserts the offloaded helpers run without an event loop in their thread. It also passes against the synchronous handlers onmain, which already ran in a worker thread. Its failing control is this PR with the helpers called directly on the loop: all six cases fail.test_xcom_read_holds_no_route_connection_during_the_producer_lookup(with and without[core] multi_team) asserts no async connection is checked out while the lookup runs. With the owner check back on the route's session, the single-team case fails (1 checked out); withhas_xcom_accessback on the request session, the multi-team case fails (2).test_trigger_xcom_request_holds_no_route_connection_during_the_producer_lookup(GETandPOSTthroughInProcessExecutionAPIwith the trigger header) asserts the same for trigger requests. With the identity check back on the route's session, both cases fail (1 checked out).Validation of the current commit through Breeze, Python 3.14.10:
sqlite+aiosqlitepostgresql+psycopg_asyncmysql+aiomysqlCommand for each backend:
On the current commit, all Execution API tests and the triggerer job tests passed on SQLite (1601 passed, 14 skipped), and fast and manual prek checks passed, including Ruff, mypy and the API generation checks; no generated files changed. The broader suites ran on the previous revision, before the trigger check moved to its own session and before the latest rebase. The triggerer job and Dag processing passed on SQLite: 1158 passed, 9 skipped. Of the suites
breeze verifylists:API Alwaysnon-DB 1262 passed, the coreapi_fastapiandalwaysDB tests 6594 passed, and thecommon.compatandfabprovider tests 622 (non-DB) and 418 (DB) passed. Code that importsduckdborpolarscrashes withIllegal instructionon the development host, which lacks AVX2. Twotest_providers_managercases crashed their workers in theAPI Alwaysrun. Theexample_duckdbimport check and thefabconnection form tests were deselected from the sequential DB runs because the crash would end the run. Each of these crashes the same way onmain.Debt left
GETon a task's own XCom builds the archived-attempt 410/404 body inroutes/xcoms.py, duplicating the one in the Execution API security module. Sharing it is a separate refactor.GETcan return 410 for an archived attempt but doesn't declare it in its OpenAPI responses. Declaring it changes the generated document, which this PR keeps byte-identical.Was generative AI tooling used to co-author this PR?
Generated-by: Claude Code (Claude Opus 5.5) following the guidelines