dshemetov opened a new issue, #70880:
URL: https://github.com/apache/airflow/issues/70880
### 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 tie-breaking mechanism in the scheduler is
absent and DAGs can run in arbitrary 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)
--
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]
