Fix asset scheduling for stale DAGs (#59337)

[Jkhall81]

fix(assets): queue asset updates for stale DAGs

closes:#59337

Summary

This PR ensures that asset updates are recorded in the AssetDagRunQueue even if a DAG is currently in a "stale" state. By allowing these updates to be queued, DAGs will automatically trigger their missed runs once they are successfully parsed and become healthy again, rather than remaining idle.

Changes

• Backend: Modified AssetManager.register_asset_change in manager.py to remove is_stale filtering while maintaining is_paused checks across direct, alias, and name/URI asset references.
• Database: Refined the SQLAlchemy query for asset name/URI references to explicitly exclude paused DAGs for logic consistency.
• Tests: Added a regression test in test_manager.py that mocks a stale DAG state and verifies it is correctly added to the AssetDagRunQueue upon an asset event.

[Jkhall81]

I have verified that the logic changes in airflow/assets/manager.py successfully address the issue in #59337 by allowing asset updates to be queued for DAGs even when they are in a 'stale' state.

But there are failures in the following tests:

• test_asset_listener_on_asset_changed_gets_calls
• test_no_create_dag_runs_when_dag_disabled[active] (across Postgres, MySQL, and Sqlite)

These are negative tests designed to assert that stale DAGs are ignored. Since that behavior was changed in the PR, these assertions are now failing.

If code change in the PR is accepted, then I can reverse the assertions of these four tests so they align with the fix.

[Lee-W]

Rebase from the latest main and fix a few nitpicks. I'm going to merge it tomorrow if no one objects.

[github-actions]

Backport failed to create: v3-1-test. View the failure log Run details

Status	Branch	Result
❌	v3-1-test

You can attempt to backport this manually by running:

cherry_picker 73451c5 v3-1-test

This should apply the commit to the v3-1-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.

After you have resolved the conflicts, you can continue the backport process by running:

cherry_picker --continue

If you don't have cherry-picker installed, see the installation guide.

[Jkhall81]

fix(assets): queue asset updates for stale DAGs

closes:#59337

Summary

This PR ensures that asset updates are recorded in the AssetDagRunQueue even if a DAG is currently in a "stale" state. By allowing these updates to be queued, DAGs will automatically trigger their missed runs once they are successfully parsed and become healthy again, rather than remaining idle.

Changes

• Backend: Modified AssetManager.register_asset_change in manager.py to remove is_stale filtering while maintaining is_paused checks across direct, alias, and name/URI asset references.
• Database: Refined the SQLAlchemy query for asset name/URI references to explicitly exclude paused DAGs for logic consistency.
• Tests: Added a regression test in test_manager.py that mocks a stale DAG state and verifies it is correctly added to the AssetDagRunQueue upon an asset event.

[Jkhall81]

I have verified that the logic changes in airflow/assets/manager.py successfully address the issue in #59337 by allowing asset updates to be queued for DAGs even when they are in a 'stale' state.

But there are failures in the following tests:

• test_asset_listener_on_asset_changed_gets_calls
• test_no_create_dag_runs_when_dag_disabled[active] (across Postgres, MySQL, and Sqlite)

These are negative tests designed to assert that stale DAGs are ignored. Since that behavior was changed in the PR, these assertions are now failing.

If code change in the PR is accepted, then I can reverse the assertions of these four tests so they align with the fix.

[Lee-W]

Rebase from the latest main and fix a few nitpicks. I'm going to merge it tomorrow if no one objects.

[github-actions]

Backport failed to create: v3-1-test. View the failure log Run details

Status	Branch	Result
❌	v3-1-test

You can attempt to backport this manually by running:

cherry_picker 73451c5 v3-1-test

This should apply the commit to the v3-1-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.

After you have resolved the conflicts, you can continue the backport process by running:

cherry_picker --continue

If you don't have cherry-picker installed, see the installation guide.

[Jkhall81]

fix(assets): queue asset updates for stale DAGs

closes:#59337

Summary

This PR ensures that asset updates are recorded in the AssetDagRunQueue even if a DAG is currently in a "stale" state. By allowing these updates to be queued, DAGs will automatically trigger their missed runs once they are successfully parsed and become healthy again, rather than remaining idle.

Changes

• Backend: Modified AssetManager.register_asset_change in manager.py to remove is_stale filtering while maintaining is_paused checks across direct, alias, and name/URI asset references.
• Database: Refined the SQLAlchemy query for asset name/URI references to explicitly exclude paused DAGs for logic consistency.
• Tests: Added a regression test in test_manager.py that mocks a stale DAG state and verifies it is correctly added to the AssetDagRunQueue upon an asset event.

[Jkhall81]

I have verified that the logic changes in airflow/assets/manager.py successfully address the issue in #59337 by allowing asset updates to be queued for DAGs even when they are in a 'stale' state.

But there are failures in the following tests:

• test_asset_listener_on_asset_changed_gets_calls
• test_no_create_dag_runs_when_dag_disabled[active] (across Postgres, MySQL, and Sqlite)

These are negative tests designed to assert that stale DAGs are ignored. Since that behavior was changed in the PR, these assertions are now failing.

If code change in the PR is accepted, then I can reverse the assertions of these four tests so they align with the fix.

[Lee-W]

Rebase from the latest main and fix a few nitpicks. I'm going to merge it tomorrow if no one objects.

[github-actions]

Backport failed to create: v3-1-test. View the failure log Run details

Status	Branch	Result
❌	v3-1-test

You can attempt to backport this manually by running:

cherry_picker 73451c5 v3-1-test

This should apply the commit to the v3-1-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.

After you have resolved the conflicts, you can continue the backport process by running:

cherry_picker --continue

If you don't have cherry-picker installed, see the installation guide.

[Jkhall81]

fix(assets): queue asset updates for stale DAGs

closes:#59337

Summary

This PR ensures that asset updates are recorded in the AssetDagRunQueue even if a DAG is currently in a "stale" state. By allowing these updates to be queued, DAGs will automatically trigger their missed runs once they are successfully parsed and become healthy again, rather than remaining idle.

Changes

• Backend: Modified AssetManager.register_asset_change in manager.py to remove is_stale filtering while maintaining is_paused checks across direct, alias, and name/URI asset references.
• Database: Refined the SQLAlchemy query for asset name/URI references to explicitly exclude paused DAGs for logic consistency.
• Tests: Added a regression test in test_manager.py that mocks a stale DAG state and verifies it is correctly added to the AssetDagRunQueue upon an asset event.

[Jkhall81]

I have verified that the logic changes in airflow/assets/manager.py successfully address the issue in #59337 by allowing asset updates to be queued for DAGs even when they are in a 'stale' state.

But there are failures in the following tests:

• test_asset_listener_on_asset_changed_gets_calls
• test_no_create_dag_runs_when_dag_disabled[active] (across Postgres, MySQL, and Sqlite)

These are negative tests designed to assert that stale DAGs are ignored. Since that behavior was changed in the PR, these assertions are now failing.

If code change in the PR is accepted, then I can reverse the assertions of these four tests so they align with the fix.

[Lee-W]

Rebase from the latest main and fix a few nitpicks. I'm going to merge it tomorrow if no one objects.

[github-actions]

Backport failed to create: v3-1-test. View the failure log Run details

Status	Branch	Result
❌	v3-1-test

You can attempt to backport this manually by running:

cherry_picker 73451c5 v3-1-test

This should apply the commit to the v3-1-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.

After you have resolved the conflicts, you can continue the backport process by running:

cherry_picker --continue

If you don't have cherry-picker installed, see the installation guide.

[Jkhall81]

fix(assets): queue asset updates for stale DAGs

closes:#59337

Summary

This PR ensures that asset updates are recorded in the AssetDagRunQueue even if a DAG is currently in a "stale" state. By allowing these updates to be queued, DAGs will automatically trigger their missed runs once they are successfully parsed and become healthy again, rather than remaining idle.

Changes

• Backend: Modified AssetManager.register_asset_change in manager.py to remove is_stale filtering while maintaining is_paused checks across direct, alias, and name/URI asset references.
• Database: Refined the SQLAlchemy query for asset name/URI references to explicitly exclude paused DAGs for logic consistency.
• Tests: Added a regression test in test_manager.py that mocks a stale DAG state and verifies it is correctly added to the AssetDagRunQueue upon an asset event.

[Jkhall81]

I have verified that the logic changes in airflow/assets/manager.py successfully address the issue in #59337 by allowing asset updates to be queued for DAGs even when they are in a 'stale' state.

But there are failures in the following tests:

• test_asset_listener_on_asset_changed_gets_calls
• test_no_create_dag_runs_when_dag_disabled[active] (across Postgres, MySQL, and Sqlite)

These are negative tests designed to assert that stale DAGs are ignored. Since that behavior was changed in the PR, these assertions are now failing.

If code change in the PR is accepted, then I can reverse the assertions of these four tests so they align with the fix.

[Lee-W]

Rebase from the latest main and fix a few nitpicks. I'm going to merge it tomorrow if no one objects.

[github-actions]

Backport failed to create: v3-1-test. View the failure log Run details

Status	Branch	Result
❌	v3-1-test

You can attempt to backport this manually by running:

cherry_picker 73451c5 v3-1-test

This should apply the commit to the v3-1-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.

After you have resolved the conflicts, you can continue the backport process by running:

cherry_picker --continue

If you don't have cherry-picker installed, see the installation guide.

[Jkhall81]

fix(assets): queue asset updates for stale DAGs

closes:#59337

Summary

This PR ensures that asset updates are recorded in the AssetDagRunQueue even if a DAG is currently in a "stale" state. By allowing these updates to be queued, DAGs will automatically trigger their missed runs once they are successfully parsed and become healthy again, rather than remaining idle.

Changes

• Backend: Modified AssetManager.register_asset_change in manager.py to remove is_stale filtering while maintaining is_paused checks across direct, alias, and name/URI asset references.
• Database: Refined the SQLAlchemy query for asset name/URI references to explicitly exclude paused DAGs for logic consistency.
• Tests: Added a regression test in test_manager.py that mocks a stale DAG state and verifies it is correctly added to the AssetDagRunQueue upon an asset event.

[Jkhall81]

I have verified that the logic changes in airflow/assets/manager.py successfully address the issue in #59337 by allowing asset updates to be queued for DAGs even when they are in a 'stale' state.

But there are failures in the following tests:

• test_asset_listener_on_asset_changed_gets_calls
• test_no_create_dag_runs_when_dag_disabled[active] (across Postgres, MySQL, and Sqlite)

These are negative tests designed to assert that stale DAGs are ignored. Since that behavior was changed in the PR, these assertions are now failing.

If code change in the PR is accepted, then I can reverse the assertions of these four tests so they align with the fix.

[Lee-W]

Rebase from the latest main and fix a few nitpicks. I'm going to merge it tomorrow if no one objects.

[github-actions]

Backport failed to create: v3-1-test. View the failure log Run details

Status	Branch	Result
❌	v3-1-test

You can attempt to backport this manually by running:

cherry_picker 73451c5 v3-1-test

This should apply the commit to the v3-1-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.

After you have resolved the conflicts, you can continue the backport process by running:

cherry_picker --continue

If you don't have cherry-picker installed, see the installation guide.

[Jkhall81]

fix(assets): queue asset updates for stale DAGs

closes:#59337

Summary

This PR ensures that asset updates are recorded in the AssetDagRunQueue even if a DAG is currently in a "stale" state. By allowing these updates to be queued, DAGs will automatically trigger their missed runs once they are successfully parsed and become healthy again, rather than remaining idle.

Changes

• Backend: Modified AssetManager.register_asset_change in manager.py to remove is_stale filtering while maintaining is_paused checks across direct, alias, and name/URI asset references.
• Database: Refined the SQLAlchemy query for asset name/URI references to explicitly exclude paused DAGs for logic consistency.
• Tests: Added a regression test in test_manager.py that mocks a stale DAG state and verifies it is correctly added to the AssetDagRunQueue upon an asset event.

[Jkhall81]

I have verified that the logic changes in airflow/assets/manager.py successfully address the issue in #59337 by allowing asset updates to be queued for DAGs even when they are in a 'stale' state.

But there are failures in the following tests:

• test_asset_listener_on_asset_changed_gets_calls
• test_no_create_dag_runs_when_dag_disabled[active] (across Postgres, MySQL, and Sqlite)

These are negative tests designed to assert that stale DAGs are ignored. Since that behavior was changed in the PR, these assertions are now failing.

If code change in the PR is accepted, then I can reverse the assertions of these four tests so they align with the fix.

[Lee-W]

Rebase from the latest main and fix a few nitpicks. I'm going to merge it tomorrow if no one objects.

[github-actions]

Backport failed to create: v3-1-test. View the failure log Run details

Status	Branch	Result
❌	v3-1-test

You can attempt to backport this manually by running:

cherry_picker 73451c5 v3-1-test

This should apply the commit to the v3-1-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.

After you have resolved the conflicts, you can continue the backport process by running:

cherry_picker --continue

If you don't have cherry-picker installed, see the installation guide.

[Jkhall81]

fix(assets): queue asset updates for stale DAGs

closes:#59337

Summary

This PR ensures that asset updates are recorded in the AssetDagRunQueue even if a DAG is currently in a "stale" state. By allowing these updates to be queued, DAGs will automatically trigger their missed runs once they are successfully parsed and become healthy again, rather than remaining idle.

Changes

• Backend: Modified AssetManager.register_asset_change in manager.py to remove is_stale filtering while maintaining is_paused checks across direct, alias, and name/URI asset references.
• Database: Refined the SQLAlchemy query for asset name/URI references to explicitly exclude paused DAGs for logic consistency.
• Tests: Added a regression test in test_manager.py that mocks a stale DAG state and verifies it is correctly added to the AssetDagRunQueue upon an asset event.

[Jkhall81]

I have verified that the logic changes in airflow/assets/manager.py successfully address the issue in #59337 by allowing asset updates to be queued for DAGs even when they are in a 'stale' state.

But there are failures in the following tests:

• test_asset_listener_on_asset_changed_gets_calls
• test_no_create_dag_runs_when_dag_disabled[active] (across Postgres, MySQL, and Sqlite)

These are negative tests designed to assert that stale DAGs are ignored. Since that behavior was changed in the PR, these assertions are now failing.

If code change in the PR is accepted, then I can reverse the assertions of these four tests so they align with the fix.

[Lee-W]

Rebase from the latest main and fix a few nitpicks. I'm going to merge it tomorrow if no one objects.

[github-actions]

Backport failed to create: v3-1-test. View the failure log Run details

Status	Branch	Result
❌	v3-1-test

You can attempt to backport this manually by running:

cherry_picker 73451c5 v3-1-test

This should apply the commit to the v3-1-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.

After you have resolved the conflicts, you can continue the backport process by running:

cherry_picker --continue

If you don't have cherry-picker installed, see the installation guide.