Fix future DagRun rarely triggered by race conditions when max_active_runs reached its upper limit.

[doiken]

closes: #31407

---

^ 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 newsfragments.

[uranusjr]

Static check failure seems to be unrelated (also failing on main).

[eladkal]

can we have a unit test for this change to avoid regression?

[doiken]

In this PR I have changed the dag retrieval to be done with locking.
This unit test is difficult to implement and I would appreciate any ideas.

I believe the regression test is covered by the existing tests.
E.g. test_dag_file_processor_process_task_instances and test_dagrun_timeout_fails_run in test_scheduler_job.py

[doiken]

It has been a while since our last conversation,
is there anything I can do about this PR?

[nathadfield]

@doiken Just picking up the thread here, your PR still has some failed static tests.

[doiken]

@nathadfield
Thanks for your comments.
I have updated the branch and confirmed that all tests passed.

[potiuk]

Usually we don't merge changes without test, but that one would be rather difficult to write test for - LGTM

[ZackUhlenhuth]

Hello all, I seem to be running into a new deadlock issue due to this change. I've changed my dag name to DAGNAME in output below:

[�[34m2023-12-03T12:43:48.487+0000�[0m] {{�[34mscheduler_job_runner.py:�[0m1426}} INFO�[0m - DAG DAGNAME scheduling was skipped, probably because the DAG record was locked�[0m
[2023-12-03T12:43:56.335+0000] {{base.py:73}} INFO - Using connection ID 'aws_default' for task execution.
[�[34m2023-12-03T12:43:58.509+0000�[0m] {{�[34mdagrun.py:�[0m632}} ERROR�[0m - Marking run <DagRun DAGNAME @ 2023-11-26 12:00:00+00:00: scheduled__2023-11-26T12:00:00+00:00, state:running, queued_at: 2023-12-03 12:00:00.743211+00:00. externally triggered: False> failed�[0m

It seems that the scheduler fails to acquire the Dag Record lock, and then fails the entire dag as a result.

Is someone able to explain the cases under which the scheduler would fail to acquire the Dag Record lock, and if there is some configuration I can use to mitigate? I was not seeing such failures before upgrading from 2.5.1 to 2.7.2