kaxil commented on code in PR #68008:
URL: https://github.com/apache/airflow/pull/68008#discussion_r3632790647
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -3620,6 +3671,27 @@ def _purge_task_instances_without_heartbeats(
request,
)
self.executor.send_callback(request)
+
+ # This purge path leaves the executor's own "task finished but TI
still looked queued"
+ # handling in process_executor_events unreachable for this TI once
handle_failure() below
+ # moves it out of RUNNING, so the email notification has to be
sent from here directly.
+ if task is not None and task.email and (task.email_on_failure or
task.email_on_retry):
+ self.executor.send_callback(
+ EmailRequest(
+ filepath=ti.dag_model.relative_fileloc or "",
+ bundle_name=bundle_name,
+ bundle_version=bundle_version,
+ version_data=version_data,
+ ti=ti,
+ msg=msg,
+ email_type=(
+ "retry" if task_callback_type ==
TaskInstanceState.UP_FOR_RETRY else "failure"
+ ),
+ context_from_server=context_from_server,
+ )
+ )
+
+ ti.handle_failure(error=msg, session=session)
Review Comment:
@ephraimbuddy this is the row-lock path you asked to lock down, and I think
there's still a gap once a scan returns more than one TI.
`handle_failure()` here calls `save_to_db()`, which runs
[`session.commit()`](https://github.com/apache/airflow/blob/b8bcab0c76cc724deb1b84af72d449bdc73d01ec/airflow-core/src/airflow/models/taskinstance.py#L1916).
That commit releases every `FOR UPDATE` row the scan took in
[`_find_task_instances_without_heartbeats`](https://github.com/apache/airflow/blob/b8bcab0c76cc724deb1b84af72d449bdc73d01ec/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L3565),
and they're never re-acquired. `_find` and `_purge` share the one
`create_session()`, so the batch lock is meant to span both, but it only
actually survives until the first `handle_failure` commits. Iterations 2..N
then run unlocked and lean entirely on the [`refresh_from_db()`
recheck](https://github.com/apache/airflow/blob/b8bcab0c76cc724deb1b84af72d449bdc73d01ec/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L3593).
That recheck still has a TOCTOU window: with the row no longer locked, a
live-but-slow worker can commit `SUCCESS` between the recheck and the [state
overwrite in
`fetch_handle_failure_context`](https://github.com/apache/airflow/blob/b8bcab0c76cc724deb1b84af72d449bdc73d01ec/airflow-core/src/airflow/models/taskinstance.py#L1888)
(FAILED, or UP_FOR_RETRY). That's the terminal-state clobber plus spurious
failure callback the lock was added to prevent. It only bites when a genuine
zombie and a slow-but-alive TI land in the same scan batch, but that's
reachable, and the single-TI regression test can't catch it (with one TI
there's no prior commit to release the lock).
Might be worth revalidating (or re-locking) per iteration rather than
relying on the batch lock surviving the first commit.
--
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]
