fix(scheduler): Eager-load DagRun asset relationships before creating DagRunContext

[Lee-W]

Why

Timed-out Dag runs that consume asset events could fail in the scheduler with
DetachedInstanceError when building DagRunContext.

What

Ensure DagRun is reloaded with asset event relationships eagerly loaded before
timeout handling. Add a regression test covering timed-out asset-triggered Dag
runs.

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[kaxil]

Given the amount of these issues — this is probably 4th for dagrun, I wonder if we should add something defensive too in DagRunContext with a “before” pydantic validator and catch this exception and pass a None/empty list as consumed_asset_event with a warning than crashing the scheduler

[Lee-W]

Yep, probably worth investing some time in that. But I'll merge this one for now and will create an issue for that improvement.

[github-actions]

Backport successfully created: v3-1-test

Status	Branch	Result
✅	v3-1-test

[wolfier]

Closes #59693

[Lee-W]

Why

Timed-out Dag runs that consume asset events could fail in the scheduler with
DetachedInstanceError when building DagRunContext.

What

Ensure DagRun is reloaded with asset event relationships eagerly loaded before
timeout handling. Add a regression test covering timed-out asset-triggered Dag
runs.

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[kaxil]

Given the amount of these issues — this is probably 4th for dagrun, I wonder if we should add something defensive too in DagRunContext with a “before” pydantic validator and catch this exception and pass a None/empty list as consumed_asset_event with a warning than crashing the scheduler

[Lee-W]

Yep, probably worth investing some time in that. But I'll merge this one for now and will create an issue for that improvement.

[github-actions]

Backport successfully created: v3-1-test

Status	Branch	Result
✅	v3-1-test

[wolfier]

Closes #59693

[Lee-W]

Why

Timed-out Dag runs that consume asset events could fail in the scheduler with
DetachedInstanceError when building DagRunContext.

What

Ensure DagRun is reloaded with asset event relationships eagerly loaded before
timeout handling. Add a regression test covering timed-out asset-triggered Dag
runs.

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[kaxil]

Given the amount of these issues — this is probably 4th for dagrun, I wonder if we should add something defensive too in DagRunContext with a “before” pydantic validator and catch this exception and pass a None/empty list as consumed_asset_event with a warning than crashing the scheduler

[Lee-W]

Yep, probably worth investing some time in that. But I'll merge this one for now and will create an issue for that improvement.

[github-actions]

Backport successfully created: v3-1-test

Status	Branch	Result
✅	v3-1-test

[wolfier]

Closes #59693

[Lee-W]

Why

Timed-out Dag runs that consume asset events could fail in the scheduler with
DetachedInstanceError when building DagRunContext.

What

Ensure DagRun is reloaded with asset event relationships eagerly loaded before
timeout handling. Add a regression test covering timed-out asset-triggered Dag
runs.

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[kaxil]

Given the amount of these issues — this is probably 4th for dagrun, I wonder if we should add something defensive too in DagRunContext with a “before” pydantic validator and catch this exception and pass a None/empty list as consumed_asset_event with a warning than crashing the scheduler

[Lee-W]

Yep, probably worth investing some time in that. But I'll merge this one for now and will create an issue for that improvement.

[github-actions]

Backport successfully created: v3-1-test

Status	Branch	Result
✅	v3-1-test

[wolfier]

Closes #59693

[Lee-W]

Why

Timed-out Dag runs that consume asset events could fail in the scheduler with
DetachedInstanceError when building DagRunContext.

What

Ensure DagRun is reloaded with asset event relationships eagerly loaded before
timeout handling. Add a regression test covering timed-out asset-triggered Dag
runs.

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[kaxil]

Given the amount of these issues — this is probably 4th for dagrun, I wonder if we should add something defensive too in DagRunContext with a “before” pydantic validator and catch this exception and pass a None/empty list as consumed_asset_event with a warning than crashing the scheduler

[Lee-W]

Yep, probably worth investing some time in that. But I'll merge this one for now and will create an issue for that improvement.

[github-actions]

Backport successfully created: v3-1-test

Status	Branch	Result
✅	v3-1-test

[wolfier]

Closes #59693

[Lee-W]

Why

Timed-out Dag runs that consume asset events could fail in the scheduler with
DetachedInstanceError when building DagRunContext.

What

Ensure DagRun is reloaded with asset event relationships eagerly loaded before
timeout handling. Add a regression test covering timed-out asset-triggered Dag
runs.

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[kaxil]

Given the amount of these issues — this is probably 4th for dagrun, I wonder if we should add something defensive too in DagRunContext with a “before” pydantic validator and catch this exception and pass a None/empty list as consumed_asset_event with a warning than crashing the scheduler

[Lee-W]

Yep, probably worth investing some time in that. But I'll merge this one for now and will create an issue for that improvement.

[github-actions]

Backport successfully created: v3-1-test

Status	Branch	Result
✅	v3-1-test

[wolfier]

Closes #59693

[Lee-W]

Why

Timed-out Dag runs that consume asset events could fail in the scheduler with
DetachedInstanceError when building DagRunContext.

What

Ensure DagRun is reloaded with asset event relationships eagerly loaded before
timeout handling. Add a regression test covering timed-out asset-triggered Dag
runs.

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[kaxil]

Given the amount of these issues — this is probably 4th for dagrun, I wonder if we should add something defensive too in DagRunContext with a “before” pydantic validator and catch this exception and pass a None/empty list as consumed_asset_event with a warning than crashing the scheduler

[Lee-W]

Yep, probably worth investing some time in that. But I'll merge this one for now and will create an issue for that improvement.

[github-actions]

Backport successfully created: v3-1-test

Status	Branch	Result
✅	v3-1-test

[wolfier]

Closes #59693

[Lee-W]

Why

Timed-out Dag runs that consume asset events could fail in the scheduler with
DetachedInstanceError when building DagRunContext.

What

Ensure DagRun is reloaded with asset event relationships eagerly loaded before
timeout handling. Add a regression test covering timed-out asset-triggered Dag
runs.

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[kaxil]

Given the amount of these issues — this is probably 4th for dagrun, I wonder if we should add something defensive too in DagRunContext with a “before” pydantic validator and catch this exception and pass a None/empty list as consumed_asset_event with a warning than crashing the scheduler

[Lee-W]

Yep, probably worth investing some time in that. But I'll merge this one for now and will create an issue for that improvement.

[github-actions]

Backport successfully created: v3-1-test

Status	Branch	Result
✅	v3-1-test

[wolfier]

Closes #59693