KyleJamesWalker opened a new pull request, #65351:
URL: https://github.com/apache/airflow/pull/65351
## fix(dag.test): use non-scoped session for QUEUED state in _run_task
dag.test() runs the in-process execution API via a2wsgi ASGIMiddleware in a
separate thread. The API creates non-scoped sessions
(create_session(scoped=False)) for each request. When _run_task sets the task
instance to QUEUED via provide_session (which uses the scoped session), the
committed state may not be visible to the API thread due to session isolation
across thread boundaries.
This causes the API to read stale state (e.g. RUNNING from a previous task)
and raise TaskAlreadyRunningError on every task after the first.
Fix by using create_session(scoped=False) to match the API session behavior,
ensuring the QUEUED state is committed on a connection with the same isolation
characteristics as what the API reads from.
To get DAG testing in VSCode working on Airflow 3.2 I have to patch via:
```python
def _patch_dag_test_commit(): # pragma: no cover
"""Patch _run_task to commit QUEUED state before the in-process API
reads it.
Airflow 3.2.0 bug: _run_task calls ti.set_state(QUEUED) which flushes to
a
NEW_SESSION but never commits. The in-process API server opens its own
session
and still sees the previous committed state (RUNNING), causing
TaskAlreadyRunningError on every task after the first.
"""
from unittest.mock import patch as _patch
from airflow.sdk.definitions import dag as _dag_mod
_original_run_task = _dag_mod._run_task
def _patched_run_task(**kwargs):
from airflow import settings as _settings
ti = kwargs.get("ti")
if ti is not None:
session = _settings.Session()
session.commit()
return _original_run_task(**kwargs)
return _patch.object(_dag_mod, "_run_task", _patched_run_task)
def run_debug(): # pragma: no cover
"""Run the DAG in debug mode."""
import os
if os.getenv("VSCODE_DEBUGPY_ACTIVE", "false") == "true":
# Create and run the DAG in debug mode with an override.
dag = example_dag()
with _patch_dag_test_commit():
run = dag.test(
run_conf={},
)
# Gather the overview
for task_instance in run.get_task_instances():
print(f"Task {task_instance.task_id} is {task_instance.state}")
if task_instance.state == "success":
print(f" produced: {task_instance.xcom_pull()}")
if __name__ == "__main__": # pragma: no cover
run_debug()
```
---
##### Was generative AI tooling used to co-author this PR?
- [x] Yes — Claude Code (Opus 4.6) was used to co-author. All generated code
has been reviewed and understood by the author.
--
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]
