Most recent Asset events ignored when max_active_runs = 1
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 10h
- Merged PRs (30d)
- 483
Description
### Apache Airflow version
3.0.6
### If "Other Airflow 2 version" selected, which one?
_No response_
### What happened?
For a DAG which allows for only one Dag run at a time (`max_active_runs=1`), if multiple events are posted created during processing of the current DagRun, only oldest one is processed while other are ignored.
Detailed scenario:
1. Create asset event 1 (timestamp: T1)
2. DAG Run 1 triggers for asset event 1 (run_after = T1)
3. While DAG Run 1 is running: Create asset events 2, 3, and 4 (timestamps: T2, T3, T4)
4. DAG Run 1 completes (consumes only event 1)
5. DAG Run 2 triggers for asset event 2 (run_after = T2) consuming only event 2
6. DAG Run 2 completes
12. No further DAG runs are triggered automatically by the scgheduler
13. Events 3 and 4 are lost forever unless a new asset event (event 5) is created
14. If event 5 is created (timestamp: T5) DAG Run 3 is triggered consuming events 3, 4 and 5
16. Result: Events 3, 4, and 5 are all processed together in DAG Run 3
Timeline:
```
T1=10:00 T2=10:01 T3=10:02 T4=10:03 T5=10:10
│ │ │ │ │
Event1 Event2 Event3 Event4 Event5
│ │ │ │ │
│ │ └─────────┴─────────────► │
│ │ (skipped by DAG Run 2) │
│ │ │
└─►DR1 └─►DR2 └─►DR3
(E1) (E2 only) (E3,E4,E5)
```
Made some SQL queries to confirm that `run_after` for DagRun 2 matches timestamp of event 2 (instead of lastest event 4)
```SQL
SELECT start_date, run_after, run_id FROM dag_run WHERE dag_id = '' ORDER BY run_after DESC LIMIT 3;
"2025-09-24 11:54:00.343726+00" "2025-09-24 11:53:59.679149+00" "asset_triggered__2025-09-24T11:53:59.679149+00:00_mnhsp7QE" # DagRun 3
"2025-09-24 11:48:34.438651+00" "2025-09-24 11:45:29.587827+00" "asset_triggered__2025-09-24T11:45:29.587827+00:00_Pw3gZ43A" # DagRun 2 (run_after) matches timestamp of event 2, even though latest event is Event 4
"2025-09-24 11:44:57.051966+00" "2025-09-24 11:44:56.471468+00" "asset_triggered__2025-09-24T11:44:56.471468+00:00_jGlPAYkN" # DagRun 1
SELECT ae.id, ae.timestamp FROM asset_event ae JOIN asset a ON ae.asset_id = a.id WHERE a.name = '' ORDER BY ae.timestamp DESC LIMIT 5;
63 "2025-09-24 11:53:59.671414+00" # Event 5
62 "2025-09-24 11:45:33.58892+00" # Event 4
61 "2025-09-24 11:45:31.583102+00" # Event 3
60 "2025-09-24 11:45:29.579845+00" # Event 2
59 "2025-09-24 11:44:56.457813+00" # Event 1
```
### What you think should happen instead?
All AssetEvents posted during DagRun execution should be processed by next DagRun (eventually each event should create separate DagRun as discussed in #55956 but I guess this is not how event scheduling is designed currently).
### How to reproduce
1. Create a DAG to be scheduled based on an Asset and ensure `max_active_runs=1`
2. Create asset event 1
3. While DAG run 1 is processing asset event 1, create asset events 2 and 3
4. After DAG run 1 processing ends, verify that DAG run 2 consumed only event 2
5. After DAG run 2 processing ends, create asset event 4
6. Verify that DAG run 3 processed events 3 and 4
### Operating System
Debian GNU/Linux 12 (bookworm)
### Versions of Apache Airflow Providers
```
apache-airflow-providers-amazon==9.12.0
apache-airflow-providers-celery==3.12.2
apache-airflow-providers-cncf-kubernetes==10.7.0
apache-airflow-providers-common-compat==1.7.3
apache-airflow-providers-common-io==1.6.2
apache-airflow-providers-common-messaging==1.0.5
apache-airflow-providers-common-sql==1.27.5
apache-airflow-providers-docker==4.4.2
apache-airflow-providers-elasticsearch==6.3.2
apache-airflow-providers-fab==2.4.1
apache-airflow-providers-ftp==3.13.2
apache-airflow-providers-git==0.0.6
apache-airflow-providers-google==17.1.0
apache-airflow-providers-grpc==3.8.2
apache-airflow-providers-hashicorp==4.3.2
apache-airflow-providers-http==5.3.3
apache-airflow-providers-microsoft-azure==12.6.1
apache-airflow-providers-mysql==6.3.3
apache-airflow-providers-odbc==4.10.2
apache-airflow-providers-openlineage==2.6.1
apache-airflow-providers-postgres==6.2.3
apache-airflow-providers-redis==4.2.0
apache-airflow-providers-sendgrid==4.1.3
apache-airflow-providers-sftp==5.3.4
apache-airflow-providers-slack==9.1.4
apache-airflow-providers-smtp==2.2.0
apache-airflow-providers-snowflake==6.4.0
apache-airflow-providers-ssh==4.1.3
apache-airflow-providers-standard==1.2.0
```
### Deployment
Official Apache Airflow Helm Chart
### Deployment details
Deployment to Kubernetes cluster using official helm chart, version `1.18.0`.
### Anything else?
Possible root cause - for some reason `triggered_date_by_dag` ([code](https://github.com/apache/airflow/blob/3.0.6/airflow-core/src/airflow/models/dag.py#L2165)) contains timestamp of the oldest instead of most recent one `AssetEvent` from `AssetDagRunQueue`, which results in ignoring later events during scheduling ([code](https://github.com/apache/airflow/blob/3.0.6/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L1650)) while removing all asset for DAG from `AssetDagRunQueue` ([code](https://github.com/apache/airflow/blob/3.0.6/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L1671)).
### 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 with triggered_date_by_dag in airflow-core/src/airflow/models/dag.py around line 2165, then trace the asset scheduling flow in airflow-core/src/airflow/jobs/scheduler_job_runner.py around lines 1650 and 1671. Reproduce the max_active_runs=1 scenario with events created during a running DagRun. Done means later queued asset events are not discarded and are processed by subsequent DagRuns as described.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- backend, data-engineering
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Clearly specified
- Newbie friendliness
- 48/100