Fix asset event scheduling race condition and consistency

[dingo4dev]

This PR addresses a race condition in the Asset-based scheduling logic where AssetEvent records created by concurrent tasks can be "missed" or "orphaned" by the Scheduler.

The issue stems from the Visibility Gap between a database flush() (which generates the created_at timestamp) and the final commit() (which makes the record visible to other sessions).

This change ensures that the synchronization between the AssetEvent and the AssetDagRunQueue is atomic and that the timestamps are aligned to prevent the drift caused by performance issue.

• Manually assigns the same timestamp to both the AssetEvent and the ADRQ before the flush.
• Commit and populate the asset event record in database immediately, so that the event can be accessible by job scheduler
• Prevent creating dag by empty asset event as the asset event is trigger in prev_dag_run

related: #54659, #56750, #56749

---
config:
theme: redux-dark-color
look: neo
---
sequenceDiagram
participant J1 as Task Instance 1
participant J2 as Task Instance 2
participant JS as Job Scheduler
participant DB as Database
Note over J1,DB: T0
Note left of J1: T1
J1-->>DB: Asset Event 1 (AE1) & commit
activate J1
activate DB
Note left of J1: T2
J2-->>DB: Asset Event 2 (AE2) & commit
activate J2
activate DB
Note left of J1: T3
J2-->>-DB: Asset Dag Run Queue (ADRQ)<br/>created_at = AE2.timestamp<br/>(flush & commit)
deactivate DB
Note left of J1: T4
activate JS
activate DB
JS-->>DB: Fetch created_at from ADRQ
JS-->>DB: Fetch events (AE1, AE2)
JS->>JS: Create DAG with run_after = ADRQ created_at (AE2.timestamp)
deactivate DB
deactivate JS
Note left of J1: T5
J1-->>DB: Asset Dag Run Queue (ADRQ)<br/>created_at = AE1.timestamp<br/>(flush & commit)
deactivate J1
deactivate DB
Note left of J1: T6
activate JS
activate DB
JS-->>DB: Fetch created_at from ADRQ
JS->>JS: No event found<br/>(AE1.timestamp < prev DAG run_after)
deactivate DB
deactivate JS

Loading

---

Durability tradeoff (new failure mode)

On Postgres/MySQL the AssetEvent is committed in a short-lived independent session so it is visible to the scheduler before the producing task's transaction commits the alias association, AssetDagRunQueue rows, and TI state change. This is intentional and required to fix the scheduling race. The tradeoff is that a failure after the event is committed but before the caller's transaction commits can leave:

• an orphaned asset event — persisted with no AssetDagRunQueue row for the scheduler to consume, or
• a duplicate asset event — if the producing task retries register_asset_change.

Operators should be aware of this new failure mode when diagnosing missing or duplicate asset-triggered Dag runs. This is also documented in the newsfragment for this change.

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)
  Copilot

---

---

Important

🛠️ Maintainer triage note for @dingo4dev · by @potiuk · 2026-06-17 14:51 UTC

Helpful heads-up from the maintainers — please address before this PR can be reviewed:

• Static / docs checks failing (CI image checks / Static checks). Run them locally with prek run --all-files (or pre-commit run --all-files) and push the fixes.
• Failing test jobs: MySQL tests: core / DB-core:MySQL:8.0:3.10:Core...Serialization. Reproduce and fix locally, then push.
• See the Pull Request quality criteria.

The ball is in your court — you've been assigned to this PR. Fix the above, then mark it Ready for review.

_{Automated triage — may be imperfect; a maintainer takes the next look.}

[Copilot]

Pull request overview

This PR attempts to fix a race condition in asset-based scheduling where AssetEvent records created by concurrent tasks can be missed by the Scheduler due to a visibility gap between database flush() and commit() operations. The fix synchronizes timestamps between AssetEvent and AssetDagRunQueue records by manually assigning timestamps and committing the asset event immediately.

Changes:

• Modified asset event registration to commit immediately after creating AssetEvent, ensuring early visibility to the scheduler
• Updated AssetDagRunQueue creation to use the asset event's timestamp for synchronization across both PostgreSQL and slow-path implementations
• Changed scheduler logic to conditionally create DAG runs only when asset events exist, and to delete ADRQ entries more selectively based on timestamps

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 7 comments.

File	Description
airflow-core/src/airflow/jobs/scheduler_job_runner.py	Modified DAG run creation to be conditional on asset_events existence and updated ADRQ deletion to filter by timestamp
airflow-core/src/airflow/assets/manager.py	Changed from session.flush() to session.commit() for early AssetEvent persistence and updated ADRQ timestamp synchronization logic

[dingo4dev]

@Lee-W@uranusjr Please take a time to review it for me. Thanks a lot ;p

[Lee-W]

will try to take a look this week

[kaxil]

Left a few comments

[molcay]

Hi @dingo4dev, I wanted to ping you here if you missed my message on Slack.

I am repeating my message and offer for help in here as well:

Hi @Stanley Law, hope you're having a good week!

I have been following your PR, it seems great stuff and thanks for tackling this. I know bouncing between reviews and CI pipelines can take up a lot of time. If you're feeling swamped or just want someone to bounce ideas off of (since I am having trouble reproducing the issue), I'd be glad to step in and help in way that I can. If you've got it fully under control though, totally feel free to ignore this message 🙂
Just wanted to offer some support.

[dingo4dev]

Hi @molcay, Thank you so much for checking in and offering help!
Some extra eyes would be amazing. Right now I am investing the failure ci test on Mysql, the concurrent tests sometimes failure and orphan few asset event, and I can't reproduce it in my local tho.
I had logged the asset event and dag run with id and timestamp.

_______________________________________________________________TestSchedulerJob.test_create_dag_runs_when_concurrent_asset_events_created_______________________________________________________________airflow-core/tests/unit/jobs/test_scheduler_job.py:5792: intest_create_dag_runs_when_concurrent_asset_events_createdasserttotal_consumed_asset_events==ASSET_EVENT_COUNTEassert29==30--------------------------------------------Capturedstdoutsetup-----------------------------------------
] FillinguptheDagBagfrom/dev/null [airflow.dag_processing.dagbag.DagBag]
--------------------------------------------Capturedstdoutcall------------------------------------------MissingAssetEventMetadata: {(16, '2026-06-23T05:02:05.983645+00:00')}
{'event_': (8, '2026-06-23T05:02:05.967063+00:00'), 'dagrun': (434, '2026-06-23T05:02:05.989838+00:00')}
{'event_': (9, '2026-06-23T05:02:05.972418+00:00'), 'dagrun': (434, '2026-06-23T05:02:05.989838+00:00')}
{'event_': (13, '2026-06-23T05:02:05.982971+00:00'), 'dagrun': (436, '2026-06-23T05:02:05.982971+00:00')}
{'event_': (16, '2026-06-23T05:02:05.983645+00:00'), 'dagrun': ()}
{'event_': (10, '2026-06-23T05:02:05.989838+00:00'), 'dagrun': (434, '2026-06-23T05:02:05.989838+00:00')}
{'event_': (15, '2026-06-23T05:02:05.994289+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (14, '2026-06-23T05:02:05.995763+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (11, '2026-06-23T05:02:06.025005+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (12, '2026-06-23T05:02:06.034978+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (18, '2026-06-23T05:02:06.059185+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (17, '2026-06-23T05:02:06.060623+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (19, '2026-06-23T05:02:06.099119+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (20, '2026-06-23T05:02:07.001801+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (21, '2026-06-23T05:02:07.022033+00:00'), 'dagrun': (438, '2026-06-23T05:02:07.090158+00:00')}
{'event_': (22, '2026-06-23T05:02:07.083870+00:00'), 'dagrun': (438, '2026-06-23T05:02:07.090158+00:00')}
{'event_': (23, '2026-06-23T05:02:07.090158+00:00'), 'dagrun': (438, '2026-06-23T05:02:07.090158+00:00')}
{'event_': (24, '2026-06-23T05:02:07.097476+00:00'), 'dagrun': (439, '2026-06-23T05:02:08.046920+00:00')}
{'event_': (25, '2026-06-23T05:02:07.116000+00:00'), 'dagrun': (439, '2026-06-23T05:02:08.046920+00:00')}
{'event_': (26, '2026-06-23T05:02:08.046920+00:00'), 'dagrun': (439, '2026-06-23T05:02:08.046920+00:00')}
{'event_': (27, '2026-06-23T05:02:08.054838+00:00'), 'dagrun': (440, '2026-06-23T05:02:08.092453+00:00')}
{'event_': (28, '2026-06-23T05:02:08.067990+00:00'), 'dagrun': (440, '2026-06-23T05:02:08.092453+00:00')}
{'event_': (29, '2026-06-23T05:02:08.092453+00:00'), 'dagrun': (440, '2026-06-23T05:02:08.092453+00:00')}
{'event_': (30, '2026-06-23T05:02:08.094004+00:00'), 'dagrun': (441, '2026-06-23T05:02:08.125311+00:00')}
{'event_': (31, '2026-06-23T05:02:08.118823+00:00'), 'dagrun': (441, '2026-06-23T05:02:08.125311+00:00')}
{'event_': (32, '2026-06-23T05:02:08.125311+00:00'), 'dagrun': (441, '2026-06-23T05:02:08.125311+00:00')}
{'event_': (33, '2026-06-23T05:02:08.127450+00:00'), 'dagrun': (442, '2026-06-23T05:02:09.073822+00:00')}
{'event_': (34, '2026-06-23T05:02:08.154315+00:00'), 'dagrun': (442, '2026-06-23T05:02:09.073822+00:00')}
{'event_': (35, '2026-06-23T05:02:09.073822+00:00'), 'dagrun': (442, '2026-06-23T05:02:09.073822+00:00')}
{'event_': (36, '2026-06-23T05:02:09.091265+00:00'), 'dagrun': (443, '2026-06-23T05:02:09.091265+00:00')}
{'event_': (37, '2026-06-23T05:02:09.118804+00:00'), 'dagrun': (444, '2026-06-23T05:02:09.118804+00:00')}

[molcay]

Hi @molcay, Thank you so much for checking in and offering help! Some extra eyes would be amazing. Right now I am investing the failure ci test on Mysql, the concurrent tests sometimes failure and orphan few asset event, and I can't reproduce it in my local tho. I had logged the asset event and dag run with id and timestamp.

_______________________________________________________________TestSchedulerJob.test_create_dag_runs_when_concurrent_asset_events_created_______________________________________________________________airflow-core/tests/unit/jobs/test_scheduler_job.py:5792: intest_create_dag_runs_when_concurrent_asset_events_createdasserttotal_consumed_asset_events==ASSET_EVENT_COUNTEassert29==30

Hi @dingo4dev,
As I see that you already mark the PR as "ready for review" and you've already got an approval, I need to ask if you still need help on the matter?

[Lee-W]

a few nits, but overall looks good

[dingo4dev]

Hi @dingo4dev, As I see that you already mark the PR as "ready for review" and you've already got an approval, I need to ask if you still need help on the matter?

Hi @molcay ! Thanks for following up.

Even though we got an approval, I have a feeling this might not completely catch every edge case of the underlying issue. we should definitely keep a close eye on it and monitor how it behaves. Thanks again for the support. 🙌

[VladaZakharova]

Hi there!
Thank you for the work you are doing, really waiting for the changes to be merged and available <3

[Lee-W]

a few nits. mostly good

[VladaZakharova]

Looks like green and beautiful :)
Are we close to merge this change?

[Lee-W]

Hey @dingo4dev , thanks for being so patient! We can merge this after the CI is green

[dingo4dev]

This PR addresses a race condition in the Asset-based scheduling logic where AssetEvent records created by concurrent tasks can be "missed" or "orphaned" by the Scheduler.

The issue stems from the Visibility Gap between a database flush() (which generates the created_at timestamp) and the final commit() (which makes the record visible to other sessions).

This change ensures that the synchronization between the AssetEvent and the AssetDagRunQueue is atomic and that the timestamps are aligned to prevent the drift caused by performance issue.

• Manually assigns the same timestamp to both the AssetEvent and the ADRQ before the flush.
• Commit and populate the asset event record in database immediately, so that the event can be accessible by job scheduler
• Prevent creating dag by empty asset event as the asset event is trigger in prev_dag_run

related: #54659, #56750, #56749

---
config:
theme: redux-dark-color
look: neo
---
sequenceDiagram
participant J1 as Task Instance 1
participant J2 as Task Instance 2
participant JS as Job Scheduler
participant DB as Database
Note over J1,DB: T0
Note left of J1: T1
J1-->>DB: Asset Event 1 (AE1) & commit
activate J1
activate DB
Note left of J1: T2
J2-->>DB: Asset Event 2 (AE2) & commit
activate J2
activate DB
Note left of J1: T3
J2-->>-DB: Asset Dag Run Queue (ADRQ)<br/>created_at = AE2.timestamp<br/>(flush & commit)
deactivate DB
Note left of J1: T4
activate JS
activate DB
JS-->>DB: Fetch created_at from ADRQ
JS-->>DB: Fetch events (AE1, AE2)
JS->>JS: Create DAG with run_after = ADRQ created_at (AE2.timestamp)
deactivate DB
deactivate JS
Note left of J1: T5
J1-->>DB: Asset Dag Run Queue (ADRQ)<br/>created_at = AE1.timestamp<br/>(flush & commit)
deactivate J1
deactivate DB
Note left of J1: T6
activate JS
activate DB
JS-->>DB: Fetch created_at from ADRQ
JS->>JS: No event found<br/>(AE1.timestamp < prev DAG run_after)
deactivate DB
deactivate JS

Loading

---

Durability tradeoff (new failure mode)

On Postgres/MySQL the AssetEvent is committed in a short-lived independent session so it is visible to the scheduler before the producing task's transaction commits the alias association, AssetDagRunQueue rows, and TI state change. This is intentional and required to fix the scheduling race. The tradeoff is that a failure after the event is committed but before the caller's transaction commits can leave:

• an orphaned asset event — persisted with no AssetDagRunQueue row for the scheduler to consume, or
• a duplicate asset event — if the producing task retries register_asset_change.

Operators should be aware of this new failure mode when diagnosing missing or duplicate asset-triggered Dag runs. This is also documented in the newsfragment for this change.

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)
  Copilot

---

---

Important

🛠️ Maintainer triage note for @dingo4dev · by @potiuk · 2026-06-17 14:51 UTC

Helpful heads-up from the maintainers — please address before this PR can be reviewed:

• Static / docs checks failing (CI image checks / Static checks). Run them locally with prek run --all-files (or pre-commit run --all-files) and push the fixes.
• Failing test jobs: MySQL tests: core / DB-core:MySQL:8.0:3.10:Core...Serialization. Reproduce and fix locally, then push.
• See the Pull Request quality criteria.

The ball is in your court — you've been assigned to this PR. Fix the above, then mark it Ready for review.

_{Automated triage — may be imperfect; a maintainer takes the next look.}

[Copilot]

Pull request overview

This PR attempts to fix a race condition in asset-based scheduling where AssetEvent records created by concurrent tasks can be missed by the Scheduler due to a visibility gap between database flush() and commit() operations. The fix synchronizes timestamps between AssetEvent and AssetDagRunQueue records by manually assigning timestamps and committing the asset event immediately.

Changes:

• Modified asset event registration to commit immediately after creating AssetEvent, ensuring early visibility to the scheduler
• Updated AssetDagRunQueue creation to use the asset event's timestamp for synchronization across both PostgreSQL and slow-path implementations
• Changed scheduler logic to conditionally create DAG runs only when asset events exist, and to delete ADRQ entries more selectively based on timestamps

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 7 comments.

File	Description
airflow-core/src/airflow/jobs/scheduler_job_runner.py	Modified DAG run creation to be conditional on asset_events existence and updated ADRQ deletion to filter by timestamp
airflow-core/src/airflow/assets/manager.py	Changed from session.flush() to session.commit() for early AssetEvent persistence and updated ADRQ timestamp synchronization logic

[dingo4dev]

@Lee-W@uranusjr Please take a time to review it for me. Thanks a lot ;p

[Lee-W]

will try to take a look this week

[kaxil]

Left a few comments

[molcay]

Hi @dingo4dev, I wanted to ping you here if you missed my message on Slack.

I am repeating my message and offer for help in here as well:

Hi @Stanley Law, hope you're having a good week!

I have been following your PR, it seems great stuff and thanks for tackling this. I know bouncing between reviews and CI pipelines can take up a lot of time. If you're feeling swamped or just want someone to bounce ideas off of (since I am having trouble reproducing the issue), I'd be glad to step in and help in way that I can. If you've got it fully under control though, totally feel free to ignore this message 🙂
Just wanted to offer some support.

[dingo4dev]

Hi @molcay, Thank you so much for checking in and offering help!
Some extra eyes would be amazing. Right now I am investing the failure ci test on Mysql, the concurrent tests sometimes failure and orphan few asset event, and I can't reproduce it in my local tho.
I had logged the asset event and dag run with id and timestamp.

_______________________________________________________________TestSchedulerJob.test_create_dag_runs_when_concurrent_asset_events_created_______________________________________________________________airflow-core/tests/unit/jobs/test_scheduler_job.py:5792: intest_create_dag_runs_when_concurrent_asset_events_createdasserttotal_consumed_asset_events==ASSET_EVENT_COUNTEassert29==30--------------------------------------------Capturedstdoutsetup-----------------------------------------
] FillinguptheDagBagfrom/dev/null [airflow.dag_processing.dagbag.DagBag]
--------------------------------------------Capturedstdoutcall------------------------------------------MissingAssetEventMetadata: {(16, '2026-06-23T05:02:05.983645+00:00')}
{'event_': (8, '2026-06-23T05:02:05.967063+00:00'), 'dagrun': (434, '2026-06-23T05:02:05.989838+00:00')}
{'event_': (9, '2026-06-23T05:02:05.972418+00:00'), 'dagrun': (434, '2026-06-23T05:02:05.989838+00:00')}
{'event_': (13, '2026-06-23T05:02:05.982971+00:00'), 'dagrun': (436, '2026-06-23T05:02:05.982971+00:00')}
{'event_': (16, '2026-06-23T05:02:05.983645+00:00'), 'dagrun': ()}
{'event_': (10, '2026-06-23T05:02:05.989838+00:00'), 'dagrun': (434, '2026-06-23T05:02:05.989838+00:00')}
{'event_': (15, '2026-06-23T05:02:05.994289+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (14, '2026-06-23T05:02:05.995763+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (11, '2026-06-23T05:02:06.025005+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (12, '2026-06-23T05:02:06.034978+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (18, '2026-06-23T05:02:06.059185+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (17, '2026-06-23T05:02:06.060623+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (19, '2026-06-23T05:02:06.099119+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (20, '2026-06-23T05:02:07.001801+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (21, '2026-06-23T05:02:07.022033+00:00'), 'dagrun': (438, '2026-06-23T05:02:07.090158+00:00')}
{'event_': (22, '2026-06-23T05:02:07.083870+00:00'), 'dagrun': (438, '2026-06-23T05:02:07.090158+00:00')}
{'event_': (23, '2026-06-23T05:02:07.090158+00:00'), 'dagrun': (438, '2026-06-23T05:02:07.090158+00:00')}
{'event_': (24, '2026-06-23T05:02:07.097476+00:00'), 'dagrun': (439, '2026-06-23T05:02:08.046920+00:00')}
{'event_': (25, '2026-06-23T05:02:07.116000+00:00'), 'dagrun': (439, '2026-06-23T05:02:08.046920+00:00')}
{'event_': (26, '2026-06-23T05:02:08.046920+00:00'), 'dagrun': (439, '2026-06-23T05:02:08.046920+00:00')}
{'event_': (27, '2026-06-23T05:02:08.054838+00:00'), 'dagrun': (440, '2026-06-23T05:02:08.092453+00:00')}
{'event_': (28, '2026-06-23T05:02:08.067990+00:00'), 'dagrun': (440, '2026-06-23T05:02:08.092453+00:00')}
{'event_': (29, '2026-06-23T05:02:08.092453+00:00'), 'dagrun': (440, '2026-06-23T05:02:08.092453+00:00')}
{'event_': (30, '2026-06-23T05:02:08.094004+00:00'), 'dagrun': (441, '2026-06-23T05:02:08.125311+00:00')}
{'event_': (31, '2026-06-23T05:02:08.118823+00:00'), 'dagrun': (441, '2026-06-23T05:02:08.125311+00:00')}
{'event_': (32, '2026-06-23T05:02:08.125311+00:00'), 'dagrun': (441, '2026-06-23T05:02:08.125311+00:00')}
{'event_': (33, '2026-06-23T05:02:08.127450+00:00'), 'dagrun': (442, '2026-06-23T05:02:09.073822+00:00')}
{'event_': (34, '2026-06-23T05:02:08.154315+00:00'), 'dagrun': (442, '2026-06-23T05:02:09.073822+00:00')}
{'event_': (35, '2026-06-23T05:02:09.073822+00:00'), 'dagrun': (442, '2026-06-23T05:02:09.073822+00:00')}
{'event_': (36, '2026-06-23T05:02:09.091265+00:00'), 'dagrun': (443, '2026-06-23T05:02:09.091265+00:00')}
{'event_': (37, '2026-06-23T05:02:09.118804+00:00'), 'dagrun': (444, '2026-06-23T05:02:09.118804+00:00')}

[molcay]

Hi @molcay, Thank you so much for checking in and offering help! Some extra eyes would be amazing. Right now I am investing the failure ci test on Mysql, the concurrent tests sometimes failure and orphan few asset event, and I can't reproduce it in my local tho. I had logged the asset event and dag run with id and timestamp.

_______________________________________________________________TestSchedulerJob.test_create_dag_runs_when_concurrent_asset_events_created_______________________________________________________________airflow-core/tests/unit/jobs/test_scheduler_job.py:5792: intest_create_dag_runs_when_concurrent_asset_events_createdasserttotal_consumed_asset_events==ASSET_EVENT_COUNTEassert29==30

Hi @dingo4dev,
As I see that you already mark the PR as "ready for review" and you've already got an approval, I need to ask if you still need help on the matter?

[Lee-W]

a few nits, but overall looks good

[dingo4dev]

Hi @dingo4dev, As I see that you already mark the PR as "ready for review" and you've already got an approval, I need to ask if you still need help on the matter?

Hi @molcay ! Thanks for following up.

Even though we got an approval, I have a feeling this might not completely catch every edge case of the underlying issue. we should definitely keep a close eye on it and monitor how it behaves. Thanks again for the support. 🙌

[VladaZakharova]

Hi there!
Thank you for the work you are doing, really waiting for the changes to be merged and available <3

[Lee-W]

a few nits. mostly good

[VladaZakharova]

Looks like green and beautiful :)
Are we close to merge this change?

[Lee-W]

Hey @dingo4dev , thanks for being so patient! We can merge this after the CI is green

[dingo4dev]

This PR addresses a race condition in the Asset-based scheduling logic where AssetEvent records created by concurrent tasks can be "missed" or "orphaned" by the Scheduler.

The issue stems from the Visibility Gap between a database flush() (which generates the created_at timestamp) and the final commit() (which makes the record visible to other sessions).

This change ensures that the synchronization between the AssetEvent and the AssetDagRunQueue is atomic and that the timestamps are aligned to prevent the drift caused by performance issue.

• Manually assigns the same timestamp to both the AssetEvent and the ADRQ before the flush.
• Commit and populate the asset event record in database immediately, so that the event can be accessible by job scheduler
• Prevent creating dag by empty asset event as the asset event is trigger in prev_dag_run

related: #54659, #56750, #56749

---
config:
theme: redux-dark-color
look: neo
---
sequenceDiagram
participant J1 as Task Instance 1
participant J2 as Task Instance 2
participant JS as Job Scheduler
participant DB as Database
Note over J1,DB: T0
Note left of J1: T1
J1-->>DB: Asset Event 1 (AE1) & commit
activate J1
activate DB
Note left of J1: T2
J2-->>DB: Asset Event 2 (AE2) & commit
activate J2
activate DB
Note left of J1: T3
J2-->>-DB: Asset Dag Run Queue (ADRQ)<br/>created_at = AE2.timestamp<br/>(flush & commit)
deactivate DB
Note left of J1: T4
activate JS
activate DB
JS-->>DB: Fetch created_at from ADRQ
JS-->>DB: Fetch events (AE1, AE2)
JS->>JS: Create DAG with run_after = ADRQ created_at (AE2.timestamp)
deactivate DB
deactivate JS
Note left of J1: T5
J1-->>DB: Asset Dag Run Queue (ADRQ)<br/>created_at = AE1.timestamp<br/>(flush & commit)
deactivate J1
deactivate DB
Note left of J1: T6
activate JS
activate DB
JS-->>DB: Fetch created_at from ADRQ
JS->>JS: No event found<br/>(AE1.timestamp < prev DAG run_after)
deactivate DB
deactivate JS

Loading

---

Durability tradeoff (new failure mode)

On Postgres/MySQL the AssetEvent is committed in a short-lived independent session so it is visible to the scheduler before the producing task's transaction commits the alias association, AssetDagRunQueue rows, and TI state change. This is intentional and required to fix the scheduling race. The tradeoff is that a failure after the event is committed but before the caller's transaction commits can leave:

• an orphaned asset event — persisted with no AssetDagRunQueue row for the scheduler to consume, or
• a duplicate asset event — if the producing task retries register_asset_change.

Operators should be aware of this new failure mode when diagnosing missing or duplicate asset-triggered Dag runs. This is also documented in the newsfragment for this change.

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)
  Copilot

---

---

Important

🛠️ Maintainer triage note for @dingo4dev · by @potiuk · 2026-06-17 14:51 UTC

Helpful heads-up from the maintainers — please address before this PR can be reviewed:

• Static / docs checks failing (CI image checks / Static checks). Run them locally with prek run --all-files (or pre-commit run --all-files) and push the fixes.
• Failing test jobs: MySQL tests: core / DB-core:MySQL:8.0:3.10:Core...Serialization. Reproduce and fix locally, then push.
• See the Pull Request quality criteria.

The ball is in your court — you've been assigned to this PR. Fix the above, then mark it Ready for review.

_{Automated triage — may be imperfect; a maintainer takes the next look.}

[Copilot]

Pull request overview

This PR attempts to fix a race condition in asset-based scheduling where AssetEvent records created by concurrent tasks can be missed by the Scheduler due to a visibility gap between database flush() and commit() operations. The fix synchronizes timestamps between AssetEvent and AssetDagRunQueue records by manually assigning timestamps and committing the asset event immediately.

Changes:

• Modified asset event registration to commit immediately after creating AssetEvent, ensuring early visibility to the scheduler
• Updated AssetDagRunQueue creation to use the asset event's timestamp for synchronization across both PostgreSQL and slow-path implementations
• Changed scheduler logic to conditionally create DAG runs only when asset events exist, and to delete ADRQ entries more selectively based on timestamps

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 7 comments.

File	Description
airflow-core/src/airflow/jobs/scheduler_job_runner.py	Modified DAG run creation to be conditional on asset_events existence and updated ADRQ deletion to filter by timestamp
airflow-core/src/airflow/assets/manager.py	Changed from session.flush() to session.commit() for early AssetEvent persistence and updated ADRQ timestamp synchronization logic

[dingo4dev]

@Lee-W@uranusjr Please take a time to review it for me. Thanks a lot ;p

[Lee-W]

will try to take a look this week

[kaxil]

Left a few comments

[molcay]

Hi @dingo4dev, I wanted to ping you here if you missed my message on Slack.

I am repeating my message and offer for help in here as well:

Hi @Stanley Law, hope you're having a good week!

I have been following your PR, it seems great stuff and thanks for tackling this. I know bouncing between reviews and CI pipelines can take up a lot of time. If you're feeling swamped or just want someone to bounce ideas off of (since I am having trouble reproducing the issue), I'd be glad to step in and help in way that I can. If you've got it fully under control though, totally feel free to ignore this message 🙂
Just wanted to offer some support.

[dingo4dev]

Hi @molcay, Thank you so much for checking in and offering help!
Some extra eyes would be amazing. Right now I am investing the failure ci test on Mysql, the concurrent tests sometimes failure and orphan few asset event, and I can't reproduce it in my local tho.
I had logged the asset event and dag run with id and timestamp.

_______________________________________________________________TestSchedulerJob.test_create_dag_runs_when_concurrent_asset_events_created_______________________________________________________________airflow-core/tests/unit/jobs/test_scheduler_job.py:5792: intest_create_dag_runs_when_concurrent_asset_events_createdasserttotal_consumed_asset_events==ASSET_EVENT_COUNTEassert29==30--------------------------------------------Capturedstdoutsetup-----------------------------------------
] FillinguptheDagBagfrom/dev/null [airflow.dag_processing.dagbag.DagBag]
--------------------------------------------Capturedstdoutcall------------------------------------------MissingAssetEventMetadata: {(16, '2026-06-23T05:02:05.983645+00:00')}
{'event_': (8, '2026-06-23T05:02:05.967063+00:00'), 'dagrun': (434, '2026-06-23T05:02:05.989838+00:00')}
{'event_': (9, '2026-06-23T05:02:05.972418+00:00'), 'dagrun': (434, '2026-06-23T05:02:05.989838+00:00')}
{'event_': (13, '2026-06-23T05:02:05.982971+00:00'), 'dagrun': (436, '2026-06-23T05:02:05.982971+00:00')}
{'event_': (16, '2026-06-23T05:02:05.983645+00:00'), 'dagrun': ()}
{'event_': (10, '2026-06-23T05:02:05.989838+00:00'), 'dagrun': (434, '2026-06-23T05:02:05.989838+00:00')}
{'event_': (15, '2026-06-23T05:02:05.994289+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (14, '2026-06-23T05:02:05.995763+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (11, '2026-06-23T05:02:06.025005+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (12, '2026-06-23T05:02:06.034978+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (18, '2026-06-23T05:02:06.059185+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (17, '2026-06-23T05:02:06.060623+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (19, '2026-06-23T05:02:06.099119+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (20, '2026-06-23T05:02:07.001801+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (21, '2026-06-23T05:02:07.022033+00:00'), 'dagrun': (438, '2026-06-23T05:02:07.090158+00:00')}
{'event_': (22, '2026-06-23T05:02:07.083870+00:00'), 'dagrun': (438, '2026-06-23T05:02:07.090158+00:00')}
{'event_': (23, '2026-06-23T05:02:07.090158+00:00'), 'dagrun': (438, '2026-06-23T05:02:07.090158+00:00')}
{'event_': (24, '2026-06-23T05:02:07.097476+00:00'), 'dagrun': (439, '2026-06-23T05:02:08.046920+00:00')}
{'event_': (25, '2026-06-23T05:02:07.116000+00:00'), 'dagrun': (439, '2026-06-23T05:02:08.046920+00:00')}
{'event_': (26, '2026-06-23T05:02:08.046920+00:00'), 'dagrun': (439, '2026-06-23T05:02:08.046920+00:00')}
{'event_': (27, '2026-06-23T05:02:08.054838+00:00'), 'dagrun': (440, '2026-06-23T05:02:08.092453+00:00')}
{'event_': (28, '2026-06-23T05:02:08.067990+00:00'), 'dagrun': (440, '2026-06-23T05:02:08.092453+00:00')}
{'event_': (29, '2026-06-23T05:02:08.092453+00:00'), 'dagrun': (440, '2026-06-23T05:02:08.092453+00:00')}
{'event_': (30, '2026-06-23T05:02:08.094004+00:00'), 'dagrun': (441, '2026-06-23T05:02:08.125311+00:00')}
{'event_': (31, '2026-06-23T05:02:08.118823+00:00'), 'dagrun': (441, '2026-06-23T05:02:08.125311+00:00')}
{'event_': (32, '2026-06-23T05:02:08.125311+00:00'), 'dagrun': (441, '2026-06-23T05:02:08.125311+00:00')}
{'event_': (33, '2026-06-23T05:02:08.127450+00:00'), 'dagrun': (442, '2026-06-23T05:02:09.073822+00:00')}
{'event_': (34, '2026-06-23T05:02:08.154315+00:00'), 'dagrun': (442, '2026-06-23T05:02:09.073822+00:00')}
{'event_': (35, '2026-06-23T05:02:09.073822+00:00'), 'dagrun': (442, '2026-06-23T05:02:09.073822+00:00')}
{'event_': (36, '2026-06-23T05:02:09.091265+00:00'), 'dagrun': (443, '2026-06-23T05:02:09.091265+00:00')}
{'event_': (37, '2026-06-23T05:02:09.118804+00:00'), 'dagrun': (444, '2026-06-23T05:02:09.118804+00:00')}

[molcay]

Hi @molcay, Thank you so much for checking in and offering help! Some extra eyes would be amazing. Right now I am investing the failure ci test on Mysql, the concurrent tests sometimes failure and orphan few asset event, and I can't reproduce it in my local tho. I had logged the asset event and dag run with id and timestamp.

_______________________________________________________________TestSchedulerJob.test_create_dag_runs_when_concurrent_asset_events_created_______________________________________________________________airflow-core/tests/unit/jobs/test_scheduler_job.py:5792: intest_create_dag_runs_when_concurrent_asset_events_createdasserttotal_consumed_asset_events==ASSET_EVENT_COUNTEassert29==30

Hi @dingo4dev,
As I see that you already mark the PR as "ready for review" and you've already got an approval, I need to ask if you still need help on the matter?

[Lee-W]

a few nits, but overall looks good

[dingo4dev]

Hi @dingo4dev, As I see that you already mark the PR as "ready for review" and you've already got an approval, I need to ask if you still need help on the matter?

Hi @molcay ! Thanks for following up.

Even though we got an approval, I have a feeling this might not completely catch every edge case of the underlying issue. we should definitely keep a close eye on it and monitor how it behaves. Thanks again for the support. 🙌

[VladaZakharova]

Hi there!
Thank you for the work you are doing, really waiting for the changes to be merged and available <3

[Lee-W]

a few nits. mostly good

[VladaZakharova]

Looks like green and beautiful :)
Are we close to merge this change?

[Lee-W]

Hey @dingo4dev , thanks for being so patient! We can merge this after the CI is green

[dingo4dev]

This PR addresses a race condition in the Asset-based scheduling logic where AssetEvent records created by concurrent tasks can be "missed" or "orphaned" by the Scheduler.

The issue stems from the Visibility Gap between a database flush() (which generates the created_at timestamp) and the final commit() (which makes the record visible to other sessions).

This change ensures that the synchronization between the AssetEvent and the AssetDagRunQueue is atomic and that the timestamps are aligned to prevent the drift caused by performance issue.

• Manually assigns the same timestamp to both the AssetEvent and the ADRQ before the flush.
• Commit and populate the asset event record in database immediately, so that the event can be accessible by job scheduler
• Prevent creating dag by empty asset event as the asset event is trigger in prev_dag_run

related: #54659, #56750, #56749

---
config:
theme: redux-dark-color
look: neo
---
sequenceDiagram
participant J1 as Task Instance 1
participant J2 as Task Instance 2
participant JS as Job Scheduler
participant DB as Database
Note over J1,DB: T0
Note left of J1: T1
J1-->>DB: Asset Event 1 (AE1) & commit
activate J1
activate DB
Note left of J1: T2
J2-->>DB: Asset Event 2 (AE2) & commit
activate J2
activate DB
Note left of J1: T3
J2-->>-DB: Asset Dag Run Queue (ADRQ)<br/>created_at = AE2.timestamp<br/>(flush & commit)
deactivate DB
Note left of J1: T4
activate JS
activate DB
JS-->>DB: Fetch created_at from ADRQ
JS-->>DB: Fetch events (AE1, AE2)
JS->>JS: Create DAG with run_after = ADRQ created_at (AE2.timestamp)
deactivate DB
deactivate JS
Note left of J1: T5
J1-->>DB: Asset Dag Run Queue (ADRQ)<br/>created_at = AE1.timestamp<br/>(flush & commit)
deactivate J1
deactivate DB
Note left of J1: T6
activate JS
activate DB
JS-->>DB: Fetch created_at from ADRQ
JS->>JS: No event found<br/>(AE1.timestamp < prev DAG run_after)
deactivate DB
deactivate JS

Loading

---

Durability tradeoff (new failure mode)

On Postgres/MySQL the AssetEvent is committed in a short-lived independent session so it is visible to the scheduler before the producing task's transaction commits the alias association, AssetDagRunQueue rows, and TI state change. This is intentional and required to fix the scheduling race. The tradeoff is that a failure after the event is committed but before the caller's transaction commits can leave:

• an orphaned asset event — persisted with no AssetDagRunQueue row for the scheduler to consume, or
• a duplicate asset event — if the producing task retries register_asset_change.

Operators should be aware of this new failure mode when diagnosing missing or duplicate asset-triggered Dag runs. This is also documented in the newsfragment for this change.

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)
  Copilot

---

---

Important

🛠️ Maintainer triage note for @dingo4dev · by @potiuk · 2026-06-17 14:51 UTC

Helpful heads-up from the maintainers — please address before this PR can be reviewed:

• Static / docs checks failing (CI image checks / Static checks). Run them locally with prek run --all-files (or pre-commit run --all-files) and push the fixes.
• Failing test jobs: MySQL tests: core / DB-core:MySQL:8.0:3.10:Core...Serialization. Reproduce and fix locally, then push.
• See the Pull Request quality criteria.

The ball is in your court — you've been assigned to this PR. Fix the above, then mark it Ready for review.

_{Automated triage — may be imperfect; a maintainer takes the next look.}

[Copilot]

Pull request overview

This PR attempts to fix a race condition in asset-based scheduling where AssetEvent records created by concurrent tasks can be missed by the Scheduler due to a visibility gap between database flush() and commit() operations. The fix synchronizes timestamps between AssetEvent and AssetDagRunQueue records by manually assigning timestamps and committing the asset event immediately.

Changes:

• Modified asset event registration to commit immediately after creating AssetEvent, ensuring early visibility to the scheduler
• Updated AssetDagRunQueue creation to use the asset event's timestamp for synchronization across both PostgreSQL and slow-path implementations
• Changed scheduler logic to conditionally create DAG runs only when asset events exist, and to delete ADRQ entries more selectively based on timestamps

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 7 comments.

File	Description
airflow-core/src/airflow/jobs/scheduler_job_runner.py	Modified DAG run creation to be conditional on asset_events existence and updated ADRQ deletion to filter by timestamp
airflow-core/src/airflow/assets/manager.py	Changed from session.flush() to session.commit() for early AssetEvent persistence and updated ADRQ timestamp synchronization logic

[dingo4dev]

@Lee-W@uranusjr Please take a time to review it for me. Thanks a lot ;p

[Lee-W]

will try to take a look this week

[kaxil]

Left a few comments

[molcay]

Hi @dingo4dev, I wanted to ping you here if you missed my message on Slack.

I am repeating my message and offer for help in here as well:

Hi @Stanley Law, hope you're having a good week!

I have been following your PR, it seems great stuff and thanks for tackling this. I know bouncing between reviews and CI pipelines can take up a lot of time. If you're feeling swamped or just want someone to bounce ideas off of (since I am having trouble reproducing the issue), I'd be glad to step in and help in way that I can. If you've got it fully under control though, totally feel free to ignore this message 🙂
Just wanted to offer some support.

[dingo4dev]

Hi @molcay, Thank you so much for checking in and offering help!
Some extra eyes would be amazing. Right now I am investing the failure ci test on Mysql, the concurrent tests sometimes failure and orphan few asset event, and I can't reproduce it in my local tho.
I had logged the asset event and dag run with id and timestamp.

_______________________________________________________________TestSchedulerJob.test_create_dag_runs_when_concurrent_asset_events_created_______________________________________________________________airflow-core/tests/unit/jobs/test_scheduler_job.py:5792: intest_create_dag_runs_when_concurrent_asset_events_createdasserttotal_consumed_asset_events==ASSET_EVENT_COUNTEassert29==30--------------------------------------------Capturedstdoutsetup-----------------------------------------
] FillinguptheDagBagfrom/dev/null [airflow.dag_processing.dagbag.DagBag]
--------------------------------------------Capturedstdoutcall------------------------------------------MissingAssetEventMetadata: {(16, '2026-06-23T05:02:05.983645+00:00')}
{'event_': (8, '2026-06-23T05:02:05.967063+00:00'), 'dagrun': (434, '2026-06-23T05:02:05.989838+00:00')}
{'event_': (9, '2026-06-23T05:02:05.972418+00:00'), 'dagrun': (434, '2026-06-23T05:02:05.989838+00:00')}
{'event_': (13, '2026-06-23T05:02:05.982971+00:00'), 'dagrun': (436, '2026-06-23T05:02:05.982971+00:00')}
{'event_': (16, '2026-06-23T05:02:05.983645+00:00'), 'dagrun': ()}
{'event_': (10, '2026-06-23T05:02:05.989838+00:00'), 'dagrun': (434, '2026-06-23T05:02:05.989838+00:00')}
{'event_': (15, '2026-06-23T05:02:05.994289+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (14, '2026-06-23T05:02:05.995763+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (11, '2026-06-23T05:02:06.025005+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (12, '2026-06-23T05:02:06.034978+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (18, '2026-06-23T05:02:06.059185+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (17, '2026-06-23T05:02:06.060623+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (19, '2026-06-23T05:02:06.099119+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (20, '2026-06-23T05:02:07.001801+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (21, '2026-06-23T05:02:07.022033+00:00'), 'dagrun': (438, '2026-06-23T05:02:07.090158+00:00')}
{'event_': (22, '2026-06-23T05:02:07.083870+00:00'), 'dagrun': (438, '2026-06-23T05:02:07.090158+00:00')}
{'event_': (23, '2026-06-23T05:02:07.090158+00:00'), 'dagrun': (438, '2026-06-23T05:02:07.090158+00:00')}
{'event_': (24, '2026-06-23T05:02:07.097476+00:00'), 'dagrun': (439, '2026-06-23T05:02:08.046920+00:00')}
{'event_': (25, '2026-06-23T05:02:07.116000+00:00'), 'dagrun': (439, '2026-06-23T05:02:08.046920+00:00')}
{'event_': (26, '2026-06-23T05:02:08.046920+00:00'), 'dagrun': (439, '2026-06-23T05:02:08.046920+00:00')}
{'event_': (27, '2026-06-23T05:02:08.054838+00:00'), 'dagrun': (440, '2026-06-23T05:02:08.092453+00:00')}
{'event_': (28, '2026-06-23T05:02:08.067990+00:00'), 'dagrun': (440, '2026-06-23T05:02:08.092453+00:00')}
{'event_': (29, '2026-06-23T05:02:08.092453+00:00'), 'dagrun': (440, '2026-06-23T05:02:08.092453+00:00')}
{'event_': (30, '2026-06-23T05:02:08.094004+00:00'), 'dagrun': (441, '2026-06-23T05:02:08.125311+00:00')}
{'event_': (31, '2026-06-23T05:02:08.118823+00:00'), 'dagrun': (441, '2026-06-23T05:02:08.125311+00:00')}
{'event_': (32, '2026-06-23T05:02:08.125311+00:00'), 'dagrun': (441, '2026-06-23T05:02:08.125311+00:00')}
{'event_': (33, '2026-06-23T05:02:08.127450+00:00'), 'dagrun': (442, '2026-06-23T05:02:09.073822+00:00')}
{'event_': (34, '2026-06-23T05:02:08.154315+00:00'), 'dagrun': (442, '2026-06-23T05:02:09.073822+00:00')}
{'event_': (35, '2026-06-23T05:02:09.073822+00:00'), 'dagrun': (442, '2026-06-23T05:02:09.073822+00:00')}
{'event_': (36, '2026-06-23T05:02:09.091265+00:00'), 'dagrun': (443, '2026-06-23T05:02:09.091265+00:00')}
{'event_': (37, '2026-06-23T05:02:09.118804+00:00'), 'dagrun': (444, '2026-06-23T05:02:09.118804+00:00')}

[molcay]

Hi @molcay, Thank you so much for checking in and offering help! Some extra eyes would be amazing. Right now I am investing the failure ci test on Mysql, the concurrent tests sometimes failure and orphan few asset event, and I can't reproduce it in my local tho. I had logged the asset event and dag run with id and timestamp.

_______________________________________________________________TestSchedulerJob.test_create_dag_runs_when_concurrent_asset_events_created_______________________________________________________________airflow-core/tests/unit/jobs/test_scheduler_job.py:5792: intest_create_dag_runs_when_concurrent_asset_events_createdasserttotal_consumed_asset_events==ASSET_EVENT_COUNTEassert29==30

Hi @dingo4dev,
As I see that you already mark the PR as "ready for review" and you've already got an approval, I need to ask if you still need help on the matter?

[Lee-W]

a few nits, but overall looks good

[dingo4dev]

Hi @dingo4dev, As I see that you already mark the PR as "ready for review" and you've already got an approval, I need to ask if you still need help on the matter?

Hi @molcay ! Thanks for following up.

Even though we got an approval, I have a feeling this might not completely catch every edge case of the underlying issue. we should definitely keep a close eye on it and monitor how it behaves. Thanks again for the support. 🙌

[VladaZakharova]

Hi there!
Thank you for the work you are doing, really waiting for the changes to be merged and available <3

[Lee-W]

a few nits. mostly good

[VladaZakharova]

Looks like green and beautiful :)
Are we close to merge this change?

[Lee-W]

Hey @dingo4dev , thanks for being so patient! We can merge this after the CI is green

[dingo4dev]

This PR addresses a race condition in the Asset-based scheduling logic where AssetEvent records created by concurrent tasks can be "missed" or "orphaned" by the Scheduler.

The issue stems from the Visibility Gap between a database flush() (which generates the created_at timestamp) and the final commit() (which makes the record visible to other sessions).

This change ensures that the synchronization between the AssetEvent and the AssetDagRunQueue is atomic and that the timestamps are aligned to prevent the drift caused by performance issue.

• Manually assigns the same timestamp to both the AssetEvent and the ADRQ before the flush.
• Commit and populate the asset event record in database immediately, so that the event can be accessible by job scheduler
• Prevent creating dag by empty asset event as the asset event is trigger in prev_dag_run

related: #54659, #56750, #56749

---
config:
theme: redux-dark-color
look: neo
---
sequenceDiagram
participant J1 as Task Instance 1
participant J2 as Task Instance 2
participant JS as Job Scheduler
participant DB as Database
Note over J1,DB: T0
Note left of J1: T1
J1-->>DB: Asset Event 1 (AE1) & commit
activate J1
activate DB
Note left of J1: T2
J2-->>DB: Asset Event 2 (AE2) & commit
activate J2
activate DB
Note left of J1: T3
J2-->>-DB: Asset Dag Run Queue (ADRQ)<br/>created_at = AE2.timestamp<br/>(flush & commit)
deactivate DB
Note left of J1: T4
activate JS
activate DB
JS-->>DB: Fetch created_at from ADRQ
JS-->>DB: Fetch events (AE1, AE2)
JS->>JS: Create DAG with run_after = ADRQ created_at (AE2.timestamp)
deactivate DB
deactivate JS
Note left of J1: T5
J1-->>DB: Asset Dag Run Queue (ADRQ)<br/>created_at = AE1.timestamp<br/>(flush & commit)
deactivate J1
deactivate DB
Note left of J1: T6
activate JS
activate DB
JS-->>DB: Fetch created_at from ADRQ
JS->>JS: No event found<br/>(AE1.timestamp < prev DAG run_after)
deactivate DB
deactivate JS

Loading

---

Durability tradeoff (new failure mode)

On Postgres/MySQL the AssetEvent is committed in a short-lived independent session so it is visible to the scheduler before the producing task's transaction commits the alias association, AssetDagRunQueue rows, and TI state change. This is intentional and required to fix the scheduling race. The tradeoff is that a failure after the event is committed but before the caller's transaction commits can leave:

• an orphaned asset event — persisted with no AssetDagRunQueue row for the scheduler to consume, or
• a duplicate asset event — if the producing task retries register_asset_change.

Operators should be aware of this new failure mode when diagnosing missing or duplicate asset-triggered Dag runs. This is also documented in the newsfragment for this change.

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)
  Copilot

---

---

Important

🛠️ Maintainer triage note for @dingo4dev · by @potiuk · 2026-06-17 14:51 UTC

Helpful heads-up from the maintainers — please address before this PR can be reviewed:

• Static / docs checks failing (CI image checks / Static checks). Run them locally with prek run --all-files (or pre-commit run --all-files) and push the fixes.
• Failing test jobs: MySQL tests: core / DB-core:MySQL:8.0:3.10:Core...Serialization. Reproduce and fix locally, then push.
• See the Pull Request quality criteria.

The ball is in your court — you've been assigned to this PR. Fix the above, then mark it Ready for review.

_{Automated triage — may be imperfect; a maintainer takes the next look.}

[Copilot]

Pull request overview

This PR attempts to fix a race condition in asset-based scheduling where AssetEvent records created by concurrent tasks can be missed by the Scheduler due to a visibility gap between database flush() and commit() operations. The fix synchronizes timestamps between AssetEvent and AssetDagRunQueue records by manually assigning timestamps and committing the asset event immediately.

Changes:

• Modified asset event registration to commit immediately after creating AssetEvent, ensuring early visibility to the scheduler
• Updated AssetDagRunQueue creation to use the asset event's timestamp for synchronization across both PostgreSQL and slow-path implementations
• Changed scheduler logic to conditionally create DAG runs only when asset events exist, and to delete ADRQ entries more selectively based on timestamps

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 7 comments.

File	Description
airflow-core/src/airflow/jobs/scheduler_job_runner.py	Modified DAG run creation to be conditional on asset_events existence and updated ADRQ deletion to filter by timestamp
airflow-core/src/airflow/assets/manager.py	Changed from session.flush() to session.commit() for early AssetEvent persistence and updated ADRQ timestamp synchronization logic

[dingo4dev]

@Lee-W@uranusjr Please take a time to review it for me. Thanks a lot ;p

[Lee-W]

will try to take a look this week

[kaxil]

Left a few comments

[molcay]

Hi @dingo4dev, I wanted to ping you here if you missed my message on Slack.

I am repeating my message and offer for help in here as well:

Hi @Stanley Law, hope you're having a good week!

I have been following your PR, it seems great stuff and thanks for tackling this. I know bouncing between reviews and CI pipelines can take up a lot of time. If you're feeling swamped or just want someone to bounce ideas off of (since I am having trouble reproducing the issue), I'd be glad to step in and help in way that I can. If you've got it fully under control though, totally feel free to ignore this message 🙂
Just wanted to offer some support.

[dingo4dev]

Hi @molcay, Thank you so much for checking in and offering help!
Some extra eyes would be amazing. Right now I am investing the failure ci test on Mysql, the concurrent tests sometimes failure and orphan few asset event, and I can't reproduce it in my local tho.
I had logged the asset event and dag run with id and timestamp.

_______________________________________________________________TestSchedulerJob.test_create_dag_runs_when_concurrent_asset_events_created_______________________________________________________________airflow-core/tests/unit/jobs/test_scheduler_job.py:5792: intest_create_dag_runs_when_concurrent_asset_events_createdasserttotal_consumed_asset_events==ASSET_EVENT_COUNTEassert29==30--------------------------------------------Capturedstdoutsetup-----------------------------------------
] FillinguptheDagBagfrom/dev/null [airflow.dag_processing.dagbag.DagBag]
--------------------------------------------Capturedstdoutcall------------------------------------------MissingAssetEventMetadata: {(16, '2026-06-23T05:02:05.983645+00:00')}
{'event_': (8, '2026-06-23T05:02:05.967063+00:00'), 'dagrun': (434, '2026-06-23T05:02:05.989838+00:00')}
{'event_': (9, '2026-06-23T05:02:05.972418+00:00'), 'dagrun': (434, '2026-06-23T05:02:05.989838+00:00')}
{'event_': (13, '2026-06-23T05:02:05.982971+00:00'), 'dagrun': (436, '2026-06-23T05:02:05.982971+00:00')}
{'event_': (16, '2026-06-23T05:02:05.983645+00:00'), 'dagrun': ()}
{'event_': (10, '2026-06-23T05:02:05.989838+00:00'), 'dagrun': (434, '2026-06-23T05:02:05.989838+00:00')}
{'event_': (15, '2026-06-23T05:02:05.994289+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (14, '2026-06-23T05:02:05.995763+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (11, '2026-06-23T05:02:06.025005+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (12, '2026-06-23T05:02:06.034978+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (18, '2026-06-23T05:02:06.059185+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (17, '2026-06-23T05:02:06.060623+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (19, '2026-06-23T05:02:06.099119+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (20, '2026-06-23T05:02:07.001801+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (21, '2026-06-23T05:02:07.022033+00:00'), 'dagrun': (438, '2026-06-23T05:02:07.090158+00:00')}
{'event_': (22, '2026-06-23T05:02:07.083870+00:00'), 'dagrun': (438, '2026-06-23T05:02:07.090158+00:00')}
{'event_': (23, '2026-06-23T05:02:07.090158+00:00'), 'dagrun': (438, '2026-06-23T05:02:07.090158+00:00')}
{'event_': (24, '2026-06-23T05:02:07.097476+00:00'), 'dagrun': (439, '2026-06-23T05:02:08.046920+00:00')}
{'event_': (25, '2026-06-23T05:02:07.116000+00:00'), 'dagrun': (439, '2026-06-23T05:02:08.046920+00:00')}
{'event_': (26, '2026-06-23T05:02:08.046920+00:00'), 'dagrun': (439, '2026-06-23T05:02:08.046920+00:00')}
{'event_': (27, '2026-06-23T05:02:08.054838+00:00'), 'dagrun': (440, '2026-06-23T05:02:08.092453+00:00')}
{'event_': (28, '2026-06-23T05:02:08.067990+00:00'), 'dagrun': (440, '2026-06-23T05:02:08.092453+00:00')}
{'event_': (29, '2026-06-23T05:02:08.092453+00:00'), 'dagrun': (440, '2026-06-23T05:02:08.092453+00:00')}
{'event_': (30, '2026-06-23T05:02:08.094004+00:00'), 'dagrun': (441, '2026-06-23T05:02:08.125311+00:00')}
{'event_': (31, '2026-06-23T05:02:08.118823+00:00'), 'dagrun': (441, '2026-06-23T05:02:08.125311+00:00')}
{'event_': (32, '2026-06-23T05:02:08.125311+00:00'), 'dagrun': (441, '2026-06-23T05:02:08.125311+00:00')}
{'event_': (33, '2026-06-23T05:02:08.127450+00:00'), 'dagrun': (442, '2026-06-23T05:02:09.073822+00:00')}
{'event_': (34, '2026-06-23T05:02:08.154315+00:00'), 'dagrun': (442, '2026-06-23T05:02:09.073822+00:00')}
{'event_': (35, '2026-06-23T05:02:09.073822+00:00'), 'dagrun': (442, '2026-06-23T05:02:09.073822+00:00')}
{'event_': (36, '2026-06-23T05:02:09.091265+00:00'), 'dagrun': (443, '2026-06-23T05:02:09.091265+00:00')}
{'event_': (37, '2026-06-23T05:02:09.118804+00:00'), 'dagrun': (444, '2026-06-23T05:02:09.118804+00:00')}

[molcay]

Hi @molcay, Thank you so much for checking in and offering help! Some extra eyes would be amazing. Right now I am investing the failure ci test on Mysql, the concurrent tests sometimes failure and orphan few asset event, and I can't reproduce it in my local tho. I had logged the asset event and dag run with id and timestamp.

_______________________________________________________________TestSchedulerJob.test_create_dag_runs_when_concurrent_asset_events_created_______________________________________________________________airflow-core/tests/unit/jobs/test_scheduler_job.py:5792: intest_create_dag_runs_when_concurrent_asset_events_createdasserttotal_consumed_asset_events==ASSET_EVENT_COUNTEassert29==30

Hi @dingo4dev,
As I see that you already mark the PR as "ready for review" and you've already got an approval, I need to ask if you still need help on the matter?

[Lee-W]

a few nits, but overall looks good

[dingo4dev]

Hi @dingo4dev, As I see that you already mark the PR as "ready for review" and you've already got an approval, I need to ask if you still need help on the matter?

Hi @molcay ! Thanks for following up.

Even though we got an approval, I have a feeling this might not completely catch every edge case of the underlying issue. we should definitely keep a close eye on it and monitor how it behaves. Thanks again for the support. 🙌

[VladaZakharova]

Hi there!
Thank you for the work you are doing, really waiting for the changes to be merged and available <3

[Lee-W]

a few nits. mostly good

[VladaZakharova]

Looks like green and beautiful :)
Are we close to merge this change?

[Lee-W]

Hey @dingo4dev , thanks for being so patient! We can merge this after the CI is green

[dingo4dev]

This PR addresses a race condition in the Asset-based scheduling logic where AssetEvent records created by concurrent tasks can be "missed" or "orphaned" by the Scheduler.

The issue stems from the Visibility Gap between a database flush() (which generates the created_at timestamp) and the final commit() (which makes the record visible to other sessions).

This change ensures that the synchronization between the AssetEvent and the AssetDagRunQueue is atomic and that the timestamps are aligned to prevent the drift caused by performance issue.

• Manually assigns the same timestamp to both the AssetEvent and the ADRQ before the flush.
• Commit and populate the asset event record in database immediately, so that the event can be accessible by job scheduler
• Prevent creating dag by empty asset event as the asset event is trigger in prev_dag_run

related: #54659, #56750, #56749

---
config:
theme: redux-dark-color
look: neo
---
sequenceDiagram
participant J1 as Task Instance 1
participant J2 as Task Instance 2
participant JS as Job Scheduler
participant DB as Database
Note over J1,DB: T0
Note left of J1: T1
J1-->>DB: Asset Event 1 (AE1) & commit
activate J1
activate DB
Note left of J1: T2
J2-->>DB: Asset Event 2 (AE2) & commit
activate J2
activate DB
Note left of J1: T3
J2-->>-DB: Asset Dag Run Queue (ADRQ)<br/>created_at = AE2.timestamp<br/>(flush & commit)
deactivate DB
Note left of J1: T4
activate JS
activate DB
JS-->>DB: Fetch created_at from ADRQ
JS-->>DB: Fetch events (AE1, AE2)
JS->>JS: Create DAG with run_after = ADRQ created_at (AE2.timestamp)
deactivate DB
deactivate JS
Note left of J1: T5
J1-->>DB: Asset Dag Run Queue (ADRQ)<br/>created_at = AE1.timestamp<br/>(flush & commit)
deactivate J1
deactivate DB
Note left of J1: T6
activate JS
activate DB
JS-->>DB: Fetch created_at from ADRQ
JS->>JS: No event found<br/>(AE1.timestamp < prev DAG run_after)
deactivate DB
deactivate JS

Loading

---

Durability tradeoff (new failure mode)

On Postgres/MySQL the AssetEvent is committed in a short-lived independent session so it is visible to the scheduler before the producing task's transaction commits the alias association, AssetDagRunQueue rows, and TI state change. This is intentional and required to fix the scheduling race. The tradeoff is that a failure after the event is committed but before the caller's transaction commits can leave:

• an orphaned asset event — persisted with no AssetDagRunQueue row for the scheduler to consume, or
• a duplicate asset event — if the producing task retries register_asset_change.

Operators should be aware of this new failure mode when diagnosing missing or duplicate asset-triggered Dag runs. This is also documented in the newsfragment for this change.

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)
  Copilot

---

---

Important

🛠️ Maintainer triage note for @dingo4dev · by @potiuk · 2026-06-17 14:51 UTC

Helpful heads-up from the maintainers — please address before this PR can be reviewed:

• Static / docs checks failing (CI image checks / Static checks). Run them locally with prek run --all-files (or pre-commit run --all-files) and push the fixes.
• Failing test jobs: MySQL tests: core / DB-core:MySQL:8.0:3.10:Core...Serialization. Reproduce and fix locally, then push.
• See the Pull Request quality criteria.

The ball is in your court — you've been assigned to this PR. Fix the above, then mark it Ready for review.

_{Automated triage — may be imperfect; a maintainer takes the next look.}

[Copilot]

Pull request overview

This PR attempts to fix a race condition in asset-based scheduling where AssetEvent records created by concurrent tasks can be missed by the Scheduler due to a visibility gap between database flush() and commit() operations. The fix synchronizes timestamps between AssetEvent and AssetDagRunQueue records by manually assigning timestamps and committing the asset event immediately.

Changes:

• Modified asset event registration to commit immediately after creating AssetEvent, ensuring early visibility to the scheduler
• Updated AssetDagRunQueue creation to use the asset event's timestamp for synchronization across both PostgreSQL and slow-path implementations
• Changed scheduler logic to conditionally create DAG runs only when asset events exist, and to delete ADRQ entries more selectively based on timestamps

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 7 comments.

File	Description
airflow-core/src/airflow/jobs/scheduler_job_runner.py	Modified DAG run creation to be conditional on asset_events existence and updated ADRQ deletion to filter by timestamp
airflow-core/src/airflow/assets/manager.py	Changed from session.flush() to session.commit() for early AssetEvent persistence and updated ADRQ timestamp synchronization logic

[dingo4dev]

@Lee-W@uranusjr Please take a time to review it for me. Thanks a lot ;p

[Lee-W]

will try to take a look this week

[kaxil]

Left a few comments

[molcay]

Hi @dingo4dev, I wanted to ping you here if you missed my message on Slack.

I am repeating my message and offer for help in here as well:

Hi @Stanley Law, hope you're having a good week!

I have been following your PR, it seems great stuff and thanks for tackling this. I know bouncing between reviews and CI pipelines can take up a lot of time. If you're feeling swamped or just want someone to bounce ideas off of (since I am having trouble reproducing the issue), I'd be glad to step in and help in way that I can. If you've got it fully under control though, totally feel free to ignore this message 🙂
Just wanted to offer some support.

[dingo4dev]

Hi @molcay, Thank you so much for checking in and offering help!
Some extra eyes would be amazing. Right now I am investing the failure ci test on Mysql, the concurrent tests sometimes failure and orphan few asset event, and I can't reproduce it in my local tho.
I had logged the asset event and dag run with id and timestamp.

_______________________________________________________________TestSchedulerJob.test_create_dag_runs_when_concurrent_asset_events_created_______________________________________________________________airflow-core/tests/unit/jobs/test_scheduler_job.py:5792: intest_create_dag_runs_when_concurrent_asset_events_createdasserttotal_consumed_asset_events==ASSET_EVENT_COUNTEassert29==30--------------------------------------------Capturedstdoutsetup-----------------------------------------
] FillinguptheDagBagfrom/dev/null [airflow.dag_processing.dagbag.DagBag]
--------------------------------------------Capturedstdoutcall------------------------------------------MissingAssetEventMetadata: {(16, '2026-06-23T05:02:05.983645+00:00')}
{'event_': (8, '2026-06-23T05:02:05.967063+00:00'), 'dagrun': (434, '2026-06-23T05:02:05.989838+00:00')}
{'event_': (9, '2026-06-23T05:02:05.972418+00:00'), 'dagrun': (434, '2026-06-23T05:02:05.989838+00:00')}
{'event_': (13, '2026-06-23T05:02:05.982971+00:00'), 'dagrun': (436, '2026-06-23T05:02:05.982971+00:00')}
{'event_': (16, '2026-06-23T05:02:05.983645+00:00'), 'dagrun': ()}
{'event_': (10, '2026-06-23T05:02:05.989838+00:00'), 'dagrun': (434, '2026-06-23T05:02:05.989838+00:00')}
{'event_': (15, '2026-06-23T05:02:05.994289+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (14, '2026-06-23T05:02:05.995763+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (11, '2026-06-23T05:02:06.025005+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (12, '2026-06-23T05:02:06.034978+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (18, '2026-06-23T05:02:06.059185+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (17, '2026-06-23T05:02:06.060623+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (19, '2026-06-23T05:02:06.099119+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (20, '2026-06-23T05:02:07.001801+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (21, '2026-06-23T05:02:07.022033+00:00'), 'dagrun': (438, '2026-06-23T05:02:07.090158+00:00')}
{'event_': (22, '2026-06-23T05:02:07.083870+00:00'), 'dagrun': (438, '2026-06-23T05:02:07.090158+00:00')}
{'event_': (23, '2026-06-23T05:02:07.090158+00:00'), 'dagrun': (438, '2026-06-23T05:02:07.090158+00:00')}
{'event_': (24, '2026-06-23T05:02:07.097476+00:00'), 'dagrun': (439, '2026-06-23T05:02:08.046920+00:00')}
{'event_': (25, '2026-06-23T05:02:07.116000+00:00'), 'dagrun': (439, '2026-06-23T05:02:08.046920+00:00')}
{'event_': (26, '2026-06-23T05:02:08.046920+00:00'), 'dagrun': (439, '2026-06-23T05:02:08.046920+00:00')}
{'event_': (27, '2026-06-23T05:02:08.054838+00:00'), 'dagrun': (440, '2026-06-23T05:02:08.092453+00:00')}
{'event_': (28, '2026-06-23T05:02:08.067990+00:00'), 'dagrun': (440, '2026-06-23T05:02:08.092453+00:00')}
{'event_': (29, '2026-06-23T05:02:08.092453+00:00'), 'dagrun': (440, '2026-06-23T05:02:08.092453+00:00')}
{'event_': (30, '2026-06-23T05:02:08.094004+00:00'), 'dagrun': (441, '2026-06-23T05:02:08.125311+00:00')}
{'event_': (31, '2026-06-23T05:02:08.118823+00:00'), 'dagrun': (441, '2026-06-23T05:02:08.125311+00:00')}
{'event_': (32, '2026-06-23T05:02:08.125311+00:00'), 'dagrun': (441, '2026-06-23T05:02:08.125311+00:00')}
{'event_': (33, '2026-06-23T05:02:08.127450+00:00'), 'dagrun': (442, '2026-06-23T05:02:09.073822+00:00')}
{'event_': (34, '2026-06-23T05:02:08.154315+00:00'), 'dagrun': (442, '2026-06-23T05:02:09.073822+00:00')}
{'event_': (35, '2026-06-23T05:02:09.073822+00:00'), 'dagrun': (442, '2026-06-23T05:02:09.073822+00:00')}
{'event_': (36, '2026-06-23T05:02:09.091265+00:00'), 'dagrun': (443, '2026-06-23T05:02:09.091265+00:00')}
{'event_': (37, '2026-06-23T05:02:09.118804+00:00'), 'dagrun': (444, '2026-06-23T05:02:09.118804+00:00')}

[molcay]

Hi @molcay, Thank you so much for checking in and offering help! Some extra eyes would be amazing. Right now I am investing the failure ci test on Mysql, the concurrent tests sometimes failure and orphan few asset event, and I can't reproduce it in my local tho. I had logged the asset event and dag run with id and timestamp.

_______________________________________________________________TestSchedulerJob.test_create_dag_runs_when_concurrent_asset_events_created_______________________________________________________________airflow-core/tests/unit/jobs/test_scheduler_job.py:5792: intest_create_dag_runs_when_concurrent_asset_events_createdasserttotal_consumed_asset_events==ASSET_EVENT_COUNTEassert29==30

Hi @dingo4dev,
As I see that you already mark the PR as "ready for review" and you've already got an approval, I need to ask if you still need help on the matter?

[Lee-W]

a few nits, but overall looks good

[dingo4dev]

Hi @dingo4dev, As I see that you already mark the PR as "ready for review" and you've already got an approval, I need to ask if you still need help on the matter?

Hi @molcay ! Thanks for following up.

Even though we got an approval, I have a feeling this might not completely catch every edge case of the underlying issue. we should definitely keep a close eye on it and monitor how it behaves. Thanks again for the support. 🙌

[VladaZakharova]

Hi there!
Thank you for the work you are doing, really waiting for the changes to be merged and available <3

[Lee-W]

a few nits. mostly good

[VladaZakharova]

Looks like green and beautiful :)
Are we close to merge this change?

[Lee-W]

Hey @dingo4dev , thanks for being so patient! We can merge this after the CI is green

[dingo4dev]

This PR addresses a race condition in the Asset-based scheduling logic where AssetEvent records created by concurrent tasks can be "missed" or "orphaned" by the Scheduler.

The issue stems from the Visibility Gap between a database flush() (which generates the created_at timestamp) and the final commit() (which makes the record visible to other sessions).

This change ensures that the synchronization between the AssetEvent and the AssetDagRunQueue is atomic and that the timestamps are aligned to prevent the drift caused by performance issue.

• Manually assigns the same timestamp to both the AssetEvent and the ADRQ before the flush.
• Commit and populate the asset event record in database immediately, so that the event can be accessible by job scheduler
• Prevent creating dag by empty asset event as the asset event is trigger in prev_dag_run

related: #54659, #56750, #56749

---
config:
theme: redux-dark-color
look: neo
---
sequenceDiagram
participant J1 as Task Instance 1
participant J2 as Task Instance 2
participant JS as Job Scheduler
participant DB as Database
Note over J1,DB: T0
Note left of J1: T1
J1-->>DB: Asset Event 1 (AE1) & commit
activate J1
activate DB
Note left of J1: T2
J2-->>DB: Asset Event 2 (AE2) & commit
activate J2
activate DB
Note left of J1: T3
J2-->>-DB: Asset Dag Run Queue (ADRQ)<br/>created_at = AE2.timestamp<br/>(flush & commit)
deactivate DB
Note left of J1: T4
activate JS
activate DB
JS-->>DB: Fetch created_at from ADRQ
JS-->>DB: Fetch events (AE1, AE2)
JS->>JS: Create DAG with run_after = ADRQ created_at (AE2.timestamp)
deactivate DB
deactivate JS
Note left of J1: T5
J1-->>DB: Asset Dag Run Queue (ADRQ)<br/>created_at = AE1.timestamp<br/>(flush & commit)
deactivate J1
deactivate DB
Note left of J1: T6
activate JS
activate DB
JS-->>DB: Fetch created_at from ADRQ
JS->>JS: No event found<br/>(AE1.timestamp < prev DAG run_after)
deactivate DB
deactivate JS

Loading

---

Durability tradeoff (new failure mode)

On Postgres/MySQL the AssetEvent is committed in a short-lived independent session so it is visible to the scheduler before the producing task's transaction commits the alias association, AssetDagRunQueue rows, and TI state change. This is intentional and required to fix the scheduling race. The tradeoff is that a failure after the event is committed but before the caller's transaction commits can leave:

• an orphaned asset event — persisted with no AssetDagRunQueue row for the scheduler to consume, or
• a duplicate asset event — if the producing task retries register_asset_change.

Operators should be aware of this new failure mode when diagnosing missing or duplicate asset-triggered Dag runs. This is also documented in the newsfragment for this change.

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)
  Copilot

---

---

Important

🛠️ Maintainer triage note for @dingo4dev · by @potiuk · 2026-06-17 14:51 UTC

Helpful heads-up from the maintainers — please address before this PR can be reviewed:

• Static / docs checks failing (CI image checks / Static checks). Run them locally with prek run --all-files (or pre-commit run --all-files) and push the fixes.
• Failing test jobs: MySQL tests: core / DB-core:MySQL:8.0:3.10:Core...Serialization. Reproduce and fix locally, then push.
• See the Pull Request quality criteria.

The ball is in your court — you've been assigned to this PR. Fix the above, then mark it Ready for review.

_{Automated triage — may be imperfect; a maintainer takes the next look.}

[Copilot]

Pull request overview

This PR attempts to fix a race condition in asset-based scheduling where AssetEvent records created by concurrent tasks can be missed by the Scheduler due to a visibility gap between database flush() and commit() operations. The fix synchronizes timestamps between AssetEvent and AssetDagRunQueue records by manually assigning timestamps and committing the asset event immediately.

Changes:

• Modified asset event registration to commit immediately after creating AssetEvent, ensuring early visibility to the scheduler
• Updated AssetDagRunQueue creation to use the asset event's timestamp for synchronization across both PostgreSQL and slow-path implementations
• Changed scheduler logic to conditionally create DAG runs only when asset events exist, and to delete ADRQ entries more selectively based on timestamps

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 7 comments.

File	Description
airflow-core/src/airflow/jobs/scheduler_job_runner.py	Modified DAG run creation to be conditional on asset_events existence and updated ADRQ deletion to filter by timestamp
airflow-core/src/airflow/assets/manager.py	Changed from session.flush() to session.commit() for early AssetEvent persistence and updated ADRQ timestamp synchronization logic

[dingo4dev]

@Lee-W@uranusjr Please take a time to review it for me. Thanks a lot ;p

[Lee-W]

will try to take a look this week

[kaxil]

Left a few comments

[molcay]

Hi @dingo4dev, I wanted to ping you here if you missed my message on Slack.

I am repeating my message and offer for help in here as well:

Hi @Stanley Law, hope you're having a good week!

I have been following your PR, it seems great stuff and thanks for tackling this. I know bouncing between reviews and CI pipelines can take up a lot of time. If you're feeling swamped or just want someone to bounce ideas off of (since I am having trouble reproducing the issue), I'd be glad to step in and help in way that I can. If you've got it fully under control though, totally feel free to ignore this message 🙂
Just wanted to offer some support.

[dingo4dev]

Hi @molcay, Thank you so much for checking in and offering help!
Some extra eyes would be amazing. Right now I am investing the failure ci test on Mysql, the concurrent tests sometimes failure and orphan few asset event, and I can't reproduce it in my local tho.
I had logged the asset event and dag run with id and timestamp.

_______________________________________________________________TestSchedulerJob.test_create_dag_runs_when_concurrent_asset_events_created_______________________________________________________________airflow-core/tests/unit/jobs/test_scheduler_job.py:5792: intest_create_dag_runs_when_concurrent_asset_events_createdasserttotal_consumed_asset_events==ASSET_EVENT_COUNTEassert29==30--------------------------------------------Capturedstdoutsetup-----------------------------------------
] FillinguptheDagBagfrom/dev/null [airflow.dag_processing.dagbag.DagBag]
--------------------------------------------Capturedstdoutcall------------------------------------------MissingAssetEventMetadata: {(16, '2026-06-23T05:02:05.983645+00:00')}
{'event_': (8, '2026-06-23T05:02:05.967063+00:00'), 'dagrun': (434, '2026-06-23T05:02:05.989838+00:00')}
{'event_': (9, '2026-06-23T05:02:05.972418+00:00'), 'dagrun': (434, '2026-06-23T05:02:05.989838+00:00')}
{'event_': (13, '2026-06-23T05:02:05.982971+00:00'), 'dagrun': (436, '2026-06-23T05:02:05.982971+00:00')}
{'event_': (16, '2026-06-23T05:02:05.983645+00:00'), 'dagrun': ()}
{'event_': (10, '2026-06-23T05:02:05.989838+00:00'), 'dagrun': (434, '2026-06-23T05:02:05.989838+00:00')}
{'event_': (15, '2026-06-23T05:02:05.994289+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (14, '2026-06-23T05:02:05.995763+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (11, '2026-06-23T05:02:06.025005+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (12, '2026-06-23T05:02:06.034978+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (18, '2026-06-23T05:02:06.059185+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (17, '2026-06-23T05:02:06.060623+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (19, '2026-06-23T05:02:06.099119+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (20, '2026-06-23T05:02:07.001801+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (21, '2026-06-23T05:02:07.022033+00:00'), 'dagrun': (438, '2026-06-23T05:02:07.090158+00:00')}
{'event_': (22, '2026-06-23T05:02:07.083870+00:00'), 'dagrun': (438, '2026-06-23T05:02:07.090158+00:00')}
{'event_': (23, '2026-06-23T05:02:07.090158+00:00'), 'dagrun': (438, '2026-06-23T05:02:07.090158+00:00')}
{'event_': (24, '2026-06-23T05:02:07.097476+00:00'), 'dagrun': (439, '2026-06-23T05:02:08.046920+00:00')}
{'event_': (25, '2026-06-23T05:02:07.116000+00:00'), 'dagrun': (439, '2026-06-23T05:02:08.046920+00:00')}
{'event_': (26, '2026-06-23T05:02:08.046920+00:00'), 'dagrun': (439, '2026-06-23T05:02:08.046920+00:00')}
{'event_': (27, '2026-06-23T05:02:08.054838+00:00'), 'dagrun': (440, '2026-06-23T05:02:08.092453+00:00')}
{'event_': (28, '2026-06-23T05:02:08.067990+00:00'), 'dagrun': (440, '2026-06-23T05:02:08.092453+00:00')}
{'event_': (29, '2026-06-23T05:02:08.092453+00:00'), 'dagrun': (440, '2026-06-23T05:02:08.092453+00:00')}
{'event_': (30, '2026-06-23T05:02:08.094004+00:00'), 'dagrun': (441, '2026-06-23T05:02:08.125311+00:00')}
{'event_': (31, '2026-06-23T05:02:08.118823+00:00'), 'dagrun': (441, '2026-06-23T05:02:08.125311+00:00')}
{'event_': (32, '2026-06-23T05:02:08.125311+00:00'), 'dagrun': (441, '2026-06-23T05:02:08.125311+00:00')}
{'event_': (33, '2026-06-23T05:02:08.127450+00:00'), 'dagrun': (442, '2026-06-23T05:02:09.073822+00:00')}
{'event_': (34, '2026-06-23T05:02:08.154315+00:00'), 'dagrun': (442, '2026-06-23T05:02:09.073822+00:00')}
{'event_': (35, '2026-06-23T05:02:09.073822+00:00'), 'dagrun': (442, '2026-06-23T05:02:09.073822+00:00')}
{'event_': (36, '2026-06-23T05:02:09.091265+00:00'), 'dagrun': (443, '2026-06-23T05:02:09.091265+00:00')}
{'event_': (37, '2026-06-23T05:02:09.118804+00:00'), 'dagrun': (444, '2026-06-23T05:02:09.118804+00:00')}

[molcay]

Hi @molcay, Thank you so much for checking in and offering help! Some extra eyes would be amazing. Right now I am investing the failure ci test on Mysql, the concurrent tests sometimes failure and orphan few asset event, and I can't reproduce it in my local tho. I had logged the asset event and dag run with id and timestamp.

_______________________________________________________________TestSchedulerJob.test_create_dag_runs_when_concurrent_asset_events_created_______________________________________________________________airflow-core/tests/unit/jobs/test_scheduler_job.py:5792: intest_create_dag_runs_when_concurrent_asset_events_createdasserttotal_consumed_asset_events==ASSET_EVENT_COUNTEassert29==30

Hi @dingo4dev,
As I see that you already mark the PR as "ready for review" and you've already got an approval, I need to ask if you still need help on the matter?

[Lee-W]

a few nits, but overall looks good

[dingo4dev]

Hi @dingo4dev, As I see that you already mark the PR as "ready for review" and you've already got an approval, I need to ask if you still need help on the matter?

Hi @molcay ! Thanks for following up.

Even though we got an approval, I have a feeling this might not completely catch every edge case of the underlying issue. we should definitely keep a close eye on it and monitor how it behaves. Thanks again for the support. 🙌

[VladaZakharova]

Hi there!
Thank you for the work you are doing, really waiting for the changes to be merged and available <3

[Lee-W]

a few nits. mostly good

[VladaZakharova]

Looks like green and beautiful :)
Are we close to merge this change?

[Lee-W]

Hey @dingo4dev , thanks for being so patient! We can merge this after the CI is green

[dingo4dev]

This PR addresses a race condition in the Asset-based scheduling logic where AssetEvent records created by concurrent tasks can be "missed" or "orphaned" by the Scheduler.

The issue stems from the Visibility Gap between a database flush() (which generates the created_at timestamp) and the final commit() (which makes the record visible to other sessions).

This change ensures that the synchronization between the AssetEvent and the AssetDagRunQueue is atomic and that the timestamps are aligned to prevent the drift caused by performance issue.

• Manually assigns the same timestamp to both the AssetEvent and the ADRQ before the flush.
• Commit and populate the asset event record in database immediately, so that the event can be accessible by job scheduler
• Prevent creating dag by empty asset event as the asset event is trigger in prev_dag_run

related: #54659, #56750, #56749

---
config:
theme: redux-dark-color
look: neo
---
sequenceDiagram
participant J1 as Task Instance 1
participant J2 as Task Instance 2
participant JS as Job Scheduler
participant DB as Database
Note over J1,DB: T0
Note left of J1: T1
J1-->>DB: Asset Event 1 (AE1) & commit
activate J1
activate DB
Note left of J1: T2
J2-->>DB: Asset Event 2 (AE2) & commit
activate J2
activate DB
Note left of J1: T3
J2-->>-DB: Asset Dag Run Queue (ADRQ)<br/>created_at = AE2.timestamp<br/>(flush & commit)
deactivate DB
Note left of J1: T4
activate JS
activate DB
JS-->>DB: Fetch created_at from ADRQ
JS-->>DB: Fetch events (AE1, AE2)
JS->>JS: Create DAG with run_after = ADRQ created_at (AE2.timestamp)
deactivate DB
deactivate JS
Note left of J1: T5
J1-->>DB: Asset Dag Run Queue (ADRQ)<br/>created_at = AE1.timestamp<br/>(flush & commit)
deactivate J1
deactivate DB
Note left of J1: T6
activate JS
activate DB
JS-->>DB: Fetch created_at from ADRQ
JS->>JS: No event found<br/>(AE1.timestamp < prev DAG run_after)
deactivate DB
deactivate JS

Loading

---

Durability tradeoff (new failure mode)

On Postgres/MySQL the AssetEvent is committed in a short-lived independent session so it is visible to the scheduler before the producing task's transaction commits the alias association, AssetDagRunQueue rows, and TI state change. This is intentional and required to fix the scheduling race. The tradeoff is that a failure after the event is committed but before the caller's transaction commits can leave:

• an orphaned asset event — persisted with no AssetDagRunQueue row for the scheduler to consume, or
• a duplicate asset event — if the producing task retries register_asset_change.

Operators should be aware of this new failure mode when diagnosing missing or duplicate asset-triggered Dag runs. This is also documented in the newsfragment for this change.

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)
  Copilot

---

---

Important

🛠️ Maintainer triage note for @dingo4dev · by @potiuk · 2026-06-17 14:51 UTC

Helpful heads-up from the maintainers — please address before this PR can be reviewed:

• Static / docs checks failing (CI image checks / Static checks). Run them locally with prek run --all-files (or pre-commit run --all-files) and push the fixes.
• Failing test jobs: MySQL tests: core / DB-core:MySQL:8.0:3.10:Core...Serialization. Reproduce and fix locally, then push.
• See the Pull Request quality criteria.

The ball is in your court — you've been assigned to this PR. Fix the above, then mark it Ready for review.

_{Automated triage — may be imperfect; a maintainer takes the next look.}

[Copilot]

Pull request overview

This PR attempts to fix a race condition in asset-based scheduling where AssetEvent records created by concurrent tasks can be missed by the Scheduler due to a visibility gap between database flush() and commit() operations. The fix synchronizes timestamps between AssetEvent and AssetDagRunQueue records by manually assigning timestamps and committing the asset event immediately.

Changes:

• Modified asset event registration to commit immediately after creating AssetEvent, ensuring early visibility to the scheduler
• Updated AssetDagRunQueue creation to use the asset event's timestamp for synchronization across both PostgreSQL and slow-path implementations
• Changed scheduler logic to conditionally create DAG runs only when asset events exist, and to delete ADRQ entries more selectively based on timestamps

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 7 comments.

File	Description
airflow-core/src/airflow/jobs/scheduler_job_runner.py	Modified DAG run creation to be conditional on asset_events existence and updated ADRQ deletion to filter by timestamp
airflow-core/src/airflow/assets/manager.py	Changed from session.flush() to session.commit() for early AssetEvent persistence and updated ADRQ timestamp synchronization logic

[dingo4dev]

@Lee-W@uranusjr Please take a time to review it for me. Thanks a lot ;p

[Lee-W]

will try to take a look this week

[kaxil]

Left a few comments

[molcay]

Hi @dingo4dev, I wanted to ping you here if you missed my message on Slack.

I am repeating my message and offer for help in here as well:

Hi @Stanley Law, hope you're having a good week!

I have been following your PR, it seems great stuff and thanks for tackling this. I know bouncing between reviews and CI pipelines can take up a lot of time. If you're feeling swamped or just want someone to bounce ideas off of (since I am having trouble reproducing the issue), I'd be glad to step in and help in way that I can. If you've got it fully under control though, totally feel free to ignore this message 🙂
Just wanted to offer some support.

[dingo4dev]

Hi @molcay, Thank you so much for checking in and offering help!
Some extra eyes would be amazing. Right now I am investing the failure ci test on Mysql, the concurrent tests sometimes failure and orphan few asset event, and I can't reproduce it in my local tho.
I had logged the asset event and dag run with id and timestamp.

_______________________________________________________________TestSchedulerJob.test_create_dag_runs_when_concurrent_asset_events_created_______________________________________________________________airflow-core/tests/unit/jobs/test_scheduler_job.py:5792: intest_create_dag_runs_when_concurrent_asset_events_createdasserttotal_consumed_asset_events==ASSET_EVENT_COUNTEassert29==30--------------------------------------------Capturedstdoutsetup-----------------------------------------
] FillinguptheDagBagfrom/dev/null [airflow.dag_processing.dagbag.DagBag]
--------------------------------------------Capturedstdoutcall------------------------------------------MissingAssetEventMetadata: {(16, '2026-06-23T05:02:05.983645+00:00')}
{'event_': (8, '2026-06-23T05:02:05.967063+00:00'), 'dagrun': (434, '2026-06-23T05:02:05.989838+00:00')}
{'event_': (9, '2026-06-23T05:02:05.972418+00:00'), 'dagrun': (434, '2026-06-23T05:02:05.989838+00:00')}
{'event_': (13, '2026-06-23T05:02:05.982971+00:00'), 'dagrun': (436, '2026-06-23T05:02:05.982971+00:00')}
{'event_': (16, '2026-06-23T05:02:05.983645+00:00'), 'dagrun': ()}
{'event_': (10, '2026-06-23T05:02:05.989838+00:00'), 'dagrun': (434, '2026-06-23T05:02:05.989838+00:00')}
{'event_': (15, '2026-06-23T05:02:05.994289+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (14, '2026-06-23T05:02:05.995763+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (11, '2026-06-23T05:02:06.025005+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (12, '2026-06-23T05:02:06.034978+00:00'), 'dagrun': (435, '2026-06-23T05:02:06.034978+00:00')}
{'event_': (18, '2026-06-23T05:02:06.059185+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (17, '2026-06-23T05:02:06.060623+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (19, '2026-06-23T05:02:06.099119+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (20, '2026-06-23T05:02:07.001801+00:00'), 'dagrun': (437, '2026-06-23T05:02:07.001801+00:00')}
{'event_': (21, '2026-06-23T05:02:07.022033+00:00'), 'dagrun': (438, '2026-06-23T05:02:07.090158+00:00')}
{'event_': (22, '2026-06-23T05:02:07.083870+00:00'), 'dagrun': (438, '2026-06-23T05:02:07.090158+00:00')}
{'event_': (23, '2026-06-23T05:02:07.090158+00:00'), 'dagrun': (438, '2026-06-23T05:02:07.090158+00:00')}
{'event_': (24, '2026-06-23T05:02:07.097476+00:00'), 'dagrun': (439, '2026-06-23T05:02:08.046920+00:00')}
{'event_': (25, '2026-06-23T05:02:07.116000+00:00'), 'dagrun': (439, '2026-06-23T05:02:08.046920+00:00')}
{'event_': (26, '2026-06-23T05:02:08.046920+00:00'), 'dagrun': (439, '2026-06-23T05:02:08.046920+00:00')}
{'event_': (27, '2026-06-23T05:02:08.054838+00:00'), 'dagrun': (440, '2026-06-23T05:02:08.092453+00:00')}
{'event_': (28, '2026-06-23T05:02:08.067990+00:00'), 'dagrun': (440, '2026-06-23T05:02:08.092453+00:00')}
{'event_': (29, '2026-06-23T05:02:08.092453+00:00'), 'dagrun': (440, '2026-06-23T05:02:08.092453+00:00')}
{'event_': (30, '2026-06-23T05:02:08.094004+00:00'), 'dagrun': (441, '2026-06-23T05:02:08.125311+00:00')}
{'event_': (31, '2026-06-23T05:02:08.118823+00:00'), 'dagrun': (441, '2026-06-23T05:02:08.125311+00:00')}
{'event_': (32, '2026-06-23T05:02:08.125311+00:00'), 'dagrun': (441, '2026-06-23T05:02:08.125311+00:00')}
{'event_': (33, '2026-06-23T05:02:08.127450+00:00'), 'dagrun': (442, '2026-06-23T05:02:09.073822+00:00')}
{'event_': (34, '2026-06-23T05:02:08.154315+00:00'), 'dagrun': (442, '2026-06-23T05:02:09.073822+00:00')}
{'event_': (35, '2026-06-23T05:02:09.073822+00:00'), 'dagrun': (442, '2026-06-23T05:02:09.073822+00:00')}
{'event_': (36, '2026-06-23T05:02:09.091265+00:00'), 'dagrun': (443, '2026-06-23T05:02:09.091265+00:00')}
{'event_': (37, '2026-06-23T05:02:09.118804+00:00'), 'dagrun': (444, '2026-06-23T05:02:09.118804+00:00')}

[molcay]

Hi @molcay, Thank you so much for checking in and offering help! Some extra eyes would be amazing. Right now I am investing the failure ci test on Mysql, the concurrent tests sometimes failure and orphan few asset event, and I can't reproduce it in my local tho. I had logged the asset event and dag run with id and timestamp.

_______________________________________________________________TestSchedulerJob.test_create_dag_runs_when_concurrent_asset_events_created_______________________________________________________________airflow-core/tests/unit/jobs/test_scheduler_job.py:5792: intest_create_dag_runs_when_concurrent_asset_events_createdasserttotal_consumed_asset_events==ASSET_EVENT_COUNTEassert29==30

Hi @dingo4dev,
As I see that you already mark the PR as "ready for review" and you've already got an approval, I need to ask if you still need help on the matter?

[Lee-W]

a few nits, but overall looks good

[dingo4dev]

Hi @dingo4dev, As I see that you already mark the PR as "ready for review" and you've already got an approval, I need to ask if you still need help on the matter?

Hi @molcay ! Thanks for following up.

Even though we got an approval, I have a feeling this might not completely catch every edge case of the underlying issue. we should definitely keep a close eye on it and monitor how it behaves. Thanks again for the support. 🙌

[VladaZakharova]

Hi there!
Thank you for the work you are doing, really waiting for the changes to be merged and available <3

[Lee-W]

a few nits. mostly good

[VladaZakharova]

Looks like green and beautiful :)
Are we close to merge this change?

[Lee-W]

Hey @dingo4dev , thanks for being so patient! We can merge this after the CI is green