kaxil commented on code in PR #65351:
URL: https://github.com/apache/airflow/pull/65351#discussion_r3097532045
##########
task-sdk/src/airflow/sdk/definitions/dag.py:
##########
@@ -1454,9 +1454,15 @@ def _run_task(
from airflow.sdk.execution_time.supervisor import
run_task_in_process
from airflow.serialization.serialized_objects import
create_scheduler_operator
- # The API Server expects the task instance to be in QUEUED state
before
- # it is run.
- ti.set_state(TaskInstanceState.QUEUED)
+ # The API Server expects the task instance to be in QUEUED state
+ # before it is run. Use a non-scoped session so the commit is
+ # visible to the in-process execution API, which runs in a
+ # separate thread (via a2wsgi) with its own non-scoped session.
+ # Using the default scoped session can leave the QUEUED state
+ # invisible across thread/session boundaries, causing the API to
+ # see stale state (e.g. RUNNING) and raise TaskAlreadyRunningError.
+ with create_session(scoped=False) as session:
Review Comment:
This fix may be addressing a symptom rather than the root cause. The framing
in the comment doesn't quite match how SQLAlchemy sessions work: both scoped
and `NonScopedSession` share the same `engine` and connection pool, and
`create_session(...)` commits through that engine on context exit. After
commit, any new transaction in any thread sees the committed row regardless of
which session factory wrote it.
A more plausible mechanism: before the fix, `set_state`'s `@provide_session`
opened a scoped session that resolved to the same thread-local session the
outer `dag.test()` loop was using (since `settings.Session` is a
`scoped_session`). On context exit it called `session.close()` on that shared
session, which can disrupt subsequent operations on the outer session and cause
objects to expire unexpectedly.
Do you have a repro that pins down what's actually failing (SQLite vs
Postgres behavior, exact state the API sees)? The inline comment is going to be
load-bearing for future readers, so it should describe the real cause. If the
real cause is scoped-session reuse on the same thread, the fix below is also
incomplete (see next comment).
##########
task-sdk/src/airflow/sdk/definitions/dag.py:
##########
@@ -1473,9 +1479,11 @@ def _run_task(
ti.task = create_scheduler_operator(taskrun_result.ti.task)
if ti.state == TaskInstanceState.DEFERRED and isinstance(msg,
DeferTask) and run_triggerer:
- # API Server expects the task instance to be in QUEUED state
before
- # resuming from deferral.
- ti.set_state(TaskInstanceState.QUEUED)
+ # API Server expects the task instance to be in QUEUED state
+ # before resuming from deferral (same cross-thread visibility
+ # concern as above).
+ with create_session(scoped=False) as session:
+ ti.set_state(TaskInstanceState.QUEUED, session=session)
Review Comment:
If the non-scoped commit is genuinely needed so the API thread can see the
QUEUED state, this fix looks incomplete. Three other state writes in the same
`_run_task` function still use the scoped session:
- line 1478: `ti.set_state(taskrun_result.ti.state)` after the task run
- lines 1503-1505: `with create_session() as session: ti.state = SCHEDULED;
session.add(ti)` after trigger resume
- line 1512: `ti.set_state(TaskInstanceState.FAILED)` in the exception
handler
In particular, the post-run state at 1478 happens right before the next
iteration's QUEUED write and could race the same way. Are these paths safe
because of timing, or should they also switch to non-scoped?
--
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]
