apache / apache/airflow

Tasks from asset-triggered DAGs execute in arbitrary order - no tiebreaker in scheduler

Open
#70,880 3 comments 0 reactions 1 assignee Claimed by @dshemetov View on GitHub
area:core area:scheduler kind:bug
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.2.1+astro.2

### What happened and how to reproduce it?

TL;DR: When multiple Asset-triggered dagruns are awaiting their turn at a slot-limited task, the scheduler's task sort-keys are all identical, which leads to inconsistent run orders. In the example below, the priority weight is -1, the logical date is NULL, and the map_index is -1, so the task execution order is backend-dependent. For non-Asset based dagruns, the logical date will not be NULL, so it acts as the tie breaker. For partitioned Assets, there seems to be a different tie breaker (created_at, id).

#### Reproducible Example

Here's a minimal reproducible example: two DAGs in one file, an asset producer and a consumer. The consumer has a single task with a single slot per dag. We demonstrate that when multiple asset-triggered runs compete for the same task slot downstream, the current tie-breaking mechanism in the scheduler does not help to order these dagruns in a consistent order.

```python
import time

from airflow.sdk import DAG, Asset, task

demo_asset = Asset("ti_ordering_demo")

with DAG("ti_ordering_producer", schedule=None, catchup=False):
@task(outlets=[demo_asset])
def emit():
pass
emit()

with DAG("ti_ordering_consumer", schedule=[demo_asset], catchup=False):
@task(max_active_tis_per_dag=1)
def slow():
time.sleep(30)

slow()
```

Create 8 asset-triggered runs, spaced out by 5 seconds so the Assets don't get lumped together into one dagrun:

```bash
for i in $(seq 8); do airflow dags trigger ti_ordering_producer; sleep 5; done
```

Because the actual behavior will depend on your backing table, so out of order execution is not guaranteed to show up by running the above, we can instead look at the table the scheduler consults for the task ordering, by executing this against Airflow's tables:

```sql
SELECT dr.run_id,
-ti.priority_weight AS k1, dr.logical_date AS k2, ti.map_index AS k3,
dr.run_after, ti.state
FROM task_instance ti
JOIN dag_run dr ON dr.dag_id = ti.dag_id AND dr.run_id = ti.run_id
WHERE ti.dag_id = 'ti_ordering_consumer' AND ti.task_id = 'slow'
ORDER BY -ti.priority_weight, dr.logical_date, ti.map_index;
```

Here's my captured output (Airflow 3.2.1, Postgres 12.6) that shows that k1, k2, k3 are all identical:

```
run_id | k1 | k2 | k3 | run_after | state
------------------------------------------------------------+----+----+----+-------------------------------+-----------
asset_triggered__2026-07-31T19:56:24.813206+00:00_DNjRsTTH | -1 | | -1 | 2026-07-31 19:56:24.813206+00 | success
asset_triggered__2026-07-31T19:56:30.299017+00:00_O5cFLgWY | -1 | | -1 | 2026-07-31 19:56:30.299017+00 | running
asset_triggered__2026-07-31T19:56:37.634871+00:00_qkLjyojx | -1 | | -1 | 2026-07-31 19:56:37.634871+00 | scheduled
asset_triggered__2026-07-31T19:56:43.462454+00:00_VM7tcL7c | -1 | | -1 | 2026-07-31 19:56:43.462454+00 | scheduled
asset_triggered__2026-07-31T19:56:50.592718+00:00_jyodC22S | -1 | | -1 | 2026-07-31 19:56:50.592718+00 | scheduled
asset_triggered__2026-07-31T19:56:56.887122+00:00_FrUI9Mf7 | -1 | | -1 | 2026-07-31 19:56:56.887122+00 | scheduled
asset_triggered__2026-07-31T19:57:03.869070+00:00_7IXyYUe4 | -1 | | -1 | 2026-07-31 19:57:03.86907+00 | scheduled
```

#### Relevant Code

Here are some parts of the scheduler source code that seem relevant to this behavior:

- in `scheduler_job_runner.py`, the function `_executable_task_instances_to_queued`, these sites seem to be responsible for the sorting logic
- https://github.com/apache/airflow/blob/79fa2e622218138f1638fe3a225156e254b3e748/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L725
- https://github.com/apache/airflow/blob/79fa2e622218138f1638fe3a225156e254b3e748/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L753
- https://github.com/apache/airflow/blob/79fa2e622218138f1638fe3a225156e254b3e748/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L777-L781
- in the same file, this line seems to be setting logical_date=False for Asset-triggered dagruns
- https://github.com/apache/airflow/blob/79fa2e622218138f1638fe3a225156e254b3e748/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L2403
- in the same file, PartitionedAssets seem to use `(created_at, id)` as a tie-breaker
- https://github.com/apache/airflow/blob/79fa2e622218138f1638fe3a225156e254b3e748/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L2249
- in `models/dagrun.py`, it seems that `run_after` is `nullable=False`, so it might be a good candidate as a tie breaker (second link shows it being used to sort execution at the dagrun level)
- https://github.com/apache/airflow/blob/79fa2e622218138f1638fe3a225156e254b3e748/airflow-core/src/airflow/models/dagrun.py#L274
- https://github.com/apache/airflow/blob/79fa2e622218138f1638fe3a225156e254b3e748/airflow-core/src/airflow/models/dagrun.py#L764-L768
- in `tests/unit/jobs/test_scheduler_job.py`, some relevant tests. They don't seem to handle the case when logical_date is NULL though.
- https://github.com/apache/airflow/blob/79fa2e622218138f1638fe3a225156e254b3e748/airflow-core/tests/unit/jobs/test_scheduler_job.py#L1652
- https://github.com/apache/airflow/blob/79fa2e622218138f1638fe3a225156e254b3e748/airflow-core/tests/unit/jobs/test_scheduler_job.py#L2106
- https://github.com/apache/airflow/blob/79fa2e622218138f1638fe3a225156e254b3e748/airflow-core/tests/unit/jobs/test_scheduler_job.py#L7558

#### Possible Fix

We might be able to fix this by changing the ordering queries (in `_executable_task_instances_to_queued`) to:

```python
.order_by(-TI.priority_weight, func.coalesce(DR.logical_date, DR.run_after), DR.id, TI.map_index)
```

But it's probably best hear from people more familiar with this codebase, if what I wrote above is on track.

#### Related Issues

- #56750 — umbrella issue for asset scheduling behaviors; doesn't seem to deal with dagrun order.

#### AI Disclosure

I used Claude Code Opus 5 to help research the Airflow source code, but the above is all written by me.

### What you think should happen instead?

_No response_

### Operating System

Debian GNU/Linux 13 (trixie)

### Deployment

None

### Apache Airflow Provider(s)

_No response_

### Versions of Apache Airflow Providers

_No response_

### Official Helm Chart version

Not Applicable

### Kubernetes Version

_No response_

### Helm Chart configuration

_No response_

### Docker Image customizations

_No response_

### Anything else?

_No response_

### Are you willing to submit PR?

- [x] 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

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.