Race condition between scheduler processing events and sensor reschedule #71172

Under which category would you file this issue?

Airflow Core

Apache Airflow version

3.2.2 (checked against 3.3.0 and main, unchanged)

What happened and how to reproduce it?

A sensor running with mode="reschedule" gets marked failed by the scheduler, because the executor result for one poke is processed after the next poke has already been re-queued.

A note on the word "reschedule", since it is overloaded here: this is not the defer path. #66431 is titled "ignore stale executor success after defer reschedule", where reschedule means the scheduler re-queued a deferred TI. This is about BaseSensorOperator(mode="reschedule"). Same shape of race, different path.

Timeline of one occurrence (UTC, poke_interval=60, CeleryExecutor):

00:04:02.121  scheduler   TI queued (try_number=1, queued_by_job_id=J)
00:04:03.003  worker      Task started
00:04:07.339  worker      up_for_reschedule, next poke due 00:05:07
00:04:07.466  worker      Task finished  exit_code=0  final_state=up_for_reschedule
                          (celery task returns normally, result is SUCCESS)
00:05:33.399  scheduler   TI put back to scheduled for the next poke
00:05:36.901  scheduler   executor success from the 00:04:07 poke is drained
00:05:36.914  scheduler   ERROR state mismatch, task marked FAILED

The poke exited cleanly. Its executor success took 89s to drain, against a 60s poke interval, so by the time the scheduler looked at it the same TI was already back in scheduled for the next poke.

Executor CeleryExecutor(parallelism=512) reported that the task instance
<TaskInstance: <dag_id>.<sensor_task_id> ... [scheduled] ti_id=...>
finished with state success, but the task instance's state attribute is scheduled.

Why the current branch does not catch it:

In reschedule mode try_number does not increment between pokes, so poke N and poke N+1 share an executor key and ti_queued is True. Then ti.queued_by_job_id != job_id is False because the same scheduler handled both (with 8 schedulers it is usually a different one, which is why this is intermittent rather than constant), executor.has_task(ti) is False because the TI is back in scheduled and has not been handed to the executor yet, and the resume-after-defer condition is False because ti.next_method is None on a reschedule exit. So it falls through and the task is failed.

Not deterministically reproducible: same DAG, same sensor, most pokes are fine and occasionally one races. Anything that makes the scheduler slower to drain the event buffer widens the window. We see it on CeleryExecutor, but nothing in the mechanism is Celery-specific, and the defer variant of this family was reported on both Celery (#66374) and Local (#67287).

Checked against 3.3.0 and main. On main the resume-after-defer condition now covers QUEUED as well as SCHEDULED, but it is still gated on ti.next_method is not None, and nothing in ti_requeued references TaskReschedule.

Differences and linkages with current tickets

• Race condition between Triggerer and Scheduler #23824 / Do not fail requeued TIs #23846 (CLOSED): the original 2.x version of this race, defer path.
• Race condition between scheduler processing events and trigger completion #66374 (CLOSED): the 3.x scheduled-state variant, defer path. Fixed by fix(scheduler): ignore stale executor success after defer reschedule #66431, backported by [v3-2-test] fix(scheduler): ignore stale executor success after defer reschedule (#66431) #67089.
• Race condition between scheduler processing events and trigger completion — queued-state #67287 (CLOSED): the queued-state variant of the same defer path.
• This one: the reschedule-mode sensor path. The existing fixes are all gated on next_method is not None, which is never true for a reschedule exit, so none of them apply.

What you think should happen instead?

A stale executor success belonging to a completed reschedule poke should be treated as a requeue, the same way the defer-exit case is now, rather than as an externally killed task.

The signal that distinguishes it is a TaskReschedule row for the TI, which is what says the task exited up_for_reschedule rather than being killed. Roughly:

or (
    ti.state in (TaskInstanceState.SCHEDULED, TaskInstanceState.QUEUED)
    and state == TaskInstanceState.SUCCESS
    and ti.next_method is None
    and ti.id in ti_ids_with_reschedule
)

Without the TaskReschedule part this would also swallow genuine external kills, which is what the branch is there to catch. TaskReschedule is not referenced in scheduler_job_runner.py today and a per-event lookup would sit on the hot path, so it probably wants resolving once alongside the existing bulk TI fetch. Happy to shape it however maintainers prefer.

Operating System

Debian GNU/Linux 12 (bookworm), official apache/airflow:slim-3.2.2-python3.12 image

Deployment

Official Apache Airflow Helm Chart

Apache Airflow Provider(s)

No response

Versions of Apache Airflow Providers

apache-airflow-providers-celery==3.20.0
apache-airflow-task-sdk==1.2.2

Official Helm Chart version

1.21.0

Kubernetes Version

v1.34.2

Helm Chart configuration

CeleryExecutor, 8 scheduler replicas, core.parallelism=512, celery.worker_concurrency=16, celery.task_acks_late=false. No other scheduler-related overrides.

Docker Image customizations

Official slim image plus in-house provider packages installed with uv. No changes to airflow-core; the scheduler is stock.

Anything else?

25 occurrences over 7 weeks in one deployment, spread across about 13 different DAG authors, and every one is a sensor in mode="reschedule". It tracks scheduler and metadata database latency rather than anything in the DAGs: on our worst day, with the database CPU-saturated by an unrelated long-running query, 5 of the 25 landed within that day.

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct