Deferred task marked failed by a stale defer-exit success once the resumed attempt is already RUNNING
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 9h
- Merged PRs (30d)
- 472
Description
### Under which category would you file this issue?
Airflow Core
### Apache Airflow version
3.3.1 (checked against `main`, unchanged)
### What happened and how to reproduce it?
Another variant of the same family, on the defer path: the resumed attempt is already
`RUNNING` when the stale executor event is drained.
A deferrable sensor defers. The worker process exits cleanly on `defer()`, and the
executor reports `SUCCESS` for that try. The trigger fires, the scheduler requeues the
task instance, and the worker starts the resumed attempt — the TI reaches `RUNNING` and
clears `next_method`. Only *then* does the scheduler drain the stale executor success
event from the defer-exit. In `process_executor_events`, `ti_queued` is True (`RUNNING`
is in the tuple) and `ti_requeued` is False, so the TI is logged as `state mismatch` and
sent to `handle_failure`.
Timeline of one occurrence (UTC, single scheduler, `LocalExecutor`, several hundred
deferrable sensors in one DAG run):
04:01:22.031 TI -> deferred (worker exits; executor will report success)
04:01:23.377 trigger fired, success
04:01:25.648 TI -> running (resumed attempt starts, next_method cleared)
04:01:28.516 state mismatch (stale defer-exit success drained here)
04:01:28.520 TI -> failed
04:01:33.967 resumed worker heartbeat -> 409, process killed
Executor LocalExecutor(parallelism=32) reported that the task instance
. ... [running] ti_id=...>
finished with state success, but the task instance's state attribute is running.
Why the current branch does not catch it:
- `ti.queued_by_job_id != job_id` is False — the same scheduler handled both.
- `executor.has_task(ti)` is False — both attempts share the same `TaskInstanceKey`
(`try_number` does not increment on resume), and the defer-exit event has already
removed that key from the executor's bookkeeping.
- The resume-after-defer condition is False on two counts: it requires
`ti.state in (SCHEDULED, QUEUED)` and `ti.next_method is not None`, and by the time the
resumed attempt is running the state is `RUNNING` and `next_method` has been cleared.
So it falls through and the task is failed. The in-code comment above the guard says the
`RUNNING` case "is handled by the scheduler detecting task instances without heartbeats",
which does not apply here: the resumed attempt is alive and heartbeating normally — the
problem is the stale event, not a missing heartbeat.
With `retries: 0` on the affected tasks there is no second attempt, so a task whose
deferred work completed successfully ends as `failed` and takes the DAG run with it.
Not deterministically reproducible: anything that makes the scheduler slower to drain the
event buffer widens the window. It shows up with
- one DAG with several hundred deferrable sensors and `retries: 0`,
- an executor sharing the host with the scheduler,
- a DAG file whose import is slow enough that many workers are starting at once,
competing for the same CPU as the scheduler loop.
The deterministic fingerprint is a `state mismatch` row in `log` immediately followed by
`failed`, for a task whose trigger event was `success`.
**Differences and linkages with current tickets**
- #23824 / #23846 (CLOSED): the original 2.x version of this race, defer path.
- #66374 (CLOSED): the scheduled-state variant, defer path. Fixed by #66431, backported
by #67089.
- #67287 (CLOSED): the queued-state variant of the same defer path, fixed by #68741.
- #71172 (OPEN): the reschedule-mode sensor path. Same shape of race, different path; the
fix proposed there is gated on a `TaskReschedule` row, which never exists here.
- This one: the defer path where the resumed attempt has already reached `RUNNING`. Both
merged fixes are gated on `ti.state in (SCHEDULED, QUEUED)` **and**
`next_method is not None`, so neither applies.
### What you think should happen instead?
A stale executor success from a defer-exit should never fail a task instance that the
trigger has already resumed, regardless of whether the resumed attempt is `SCHEDULED`,
`QUEUED` or already `RUNNING`.
Widening the state tuple is not enough on its own, because `next_method` is already
cleared once the resumed attempt runs, so neither field identifies the case any more.
Two directions that would:
1. Make the stale event stop matching: bump the attempt identifier on resume so the
defer-exit event no longer matches `buffer_key` for the current attempt.
2. Compare timestamps: ignore an executor success whose TI is `RUNNING` and whose
current attempt started after the event was produced.
### Operating System
Ubuntu 22.04 (Linux 5.15)
### Deployment
Docker-Compose
### Apache Airflow Provider(s)
_No response_
### Versions of Apache Airflow Providers
Not provider-specific. Reproduced with astronomer-cosmos 1.15.1 deferrable sensors, but
the mechanism is executor/scheduler-side and provider-independent.
### Official Helm Chart version
Not Applicable
### Kubernetes Version
_No response_
### Helm Chart configuration
_No response_
### Docker Image customizations
Custom image built on the official Airflow base image; no changes to scheduler,
executor or Task SDK code.
### Anything else?
Twice in three days on the same DAG, on a different sensor each time. Our workaround is
to stop deferring those sensors, which removes the defer-exit event entirely.
### Are you willing to submit PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
Contributor guide
Research direction
Start at `process_executor_events` and inspect the existing guards for deferred and resumed task instances, including the `RUNNING` state and `next_method` checks. Use the described stale defer-exit success timeline as the reproduction target; done means a successful trigger-resumed task is not failed by that stale event, with coverage for the already-`RUNNING` case.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- backend, distributed-systems
- Issue type
- Bug
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Active
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100