[avolant]

Why

We observed many runnable task instances remaining in the SCHEDULED state for a long time. Higher-priority task instances repeatedly appeared ahead of them in the candidate query but could not be queued because of Dag-internal concurrency limits. Those blocked candidates consumed the query limit and delayed lower-priority runnable tasks.

The candidate query applies LIMIT max_tis, then runs concurrency and capacity checks in memory. On main, the selection loop stops as soon as a page yields one executable TI, or when the page is shorter than max_tis. A page dominated by blocked TIs therefore queues one TI per scheduler loop, and the same blocked TIs lead the next loop's page again.

For example, with max_tis=2 and candidates ordered as:

A: blocked by task concurrency
B: runnable
C: runnable

The first query returns [A, B]. The scheduler rejects A, accepts B, and stops with one of two positions filled. C waits for the next scheduler loop.

What changed

• The loop keeps querying while the batch is under max_tis and the previous query added a starvation filter. In the example, the second query excludes A's task through the learned filter and B by ID, returns [C], and the batch becomes [B, C].
• _build_schedulable_tis_query takes a keyword-only excluded_ti_ids, applied before the max_active_tasks row-number window like the starvation filters. Selected TIs stay SCHEDULED in the database until the loop marks them QUEUED, so later queries would otherwise return them again.
• Every query keeps LIMIT max_tis, and selection stops inside the page once the batch is full. Shrinking the limit to the remaining capacity would cost one round trip per blocked task when the batch is nearly full.
• Executor slot availability is read once before the loop, so slots taken by earlier queries stay taken.
• Queries issued after the first TI is selected are capped at MAX_TI_REFILL_QUERIES_PER_LOOP = 3. They run inside the critical section, holding the pool row locks and, on PostgreSQL, the advisory lock. Hitting the cap logs at debug level and increments the new scheduler.critical_section_refill_limit_reached counter. Queries before the first selection stay uncapped, as on main. Capping them would let a long enough run of blocked candidates starve runnable TIs indefinitely, because every scheduler loop starts with empty filters.
• A short page no longer ends the loop when it added a filter. A Dag run's max_active_tasks window can return only its blocked TIs and hide its runnable ones until the blocked TIs are filtered out.

This intentionally does not use SQL OFFSET. Each query adds newly learned filters, which changes and shrinks the ordered result set, so an offset could skip valid candidates.

Validation

New tests share one scenario where each three-row candidate page holds a single runnable TI until the last page:

• test_select_task_instances_to_queue_refills_batch_after_concurrency_rejection fills the batch to exactly max_tis when the last page has more runnable TIs than free positions. A second case with two executor slots queues exactly two TIs, so slots are shared across queries.
• test_select_task_instances_to_queue_stops_at_refill_query_limit patches the cap to 1, checks the batch stops after one refill, and checks the counter is emitted.
• test_select_task_instances_to_queue_finds_runnable_hidden_by_max_active_tasks_window covers the short-page case.

Each test fails when its invariant is removed: rebuilding executor slots per query, dropping the in-page max_tis break, removing the cap, dropping the selected-ID exclusion, or restoring the short-page exit. All four fail on main.

• Full test_scheduler_job.py on SQLite: 503 passed, 4 skipped.
• prek pre-commit and manual stages on the changed files, including mypy-airflow-core and the metrics registry check: passed.
• Not re-run on PostgreSQL after the rebase.

---

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

• Yes -- Codex (GPT-5), Claude Code (Opus 5.5)

Generated-by: Codex (GPT-5) and Claude Code (Opus 5.5) following the guidelines

[avolant]

Here is my humain's formulation of the problem:

In our production environment, we had a problem where some tasks stayed in SCHEDULED state for longer than expected. we tracked down the root cause of the problem to this line;

is_done = executable_tis or len(task_instances_to_examine) < max_tis

To our understanding, is_done condition was met (meaning, that is_done was set to True), but only one executable_tis was transitionned from SCHEDULED to QUEUED:
We had 32 as max_tis_per_query, but 31 of the studied tasks couldn't run (meaning, couldn't transition from SCHEDULED to QUEUED), because they had reached the internal DAG max_tis_per_task. The trouble is that each scheduler iterations were taking the same 31 tasks, and scheduling only task per loop which was highly inefficient, because the above is_done condition was always met.

This PR attempt to solve this by making sure the scheduler loop continues to check the next batch of schedulable tasks:

if len(executable_tis) >= max_tis or not found_new_filters:

[avolant]

Commit eac3bf9 keeps refill queries at max_tis instead of shrinking them to the remaining batch capacity. This avoids one database query per blocked task when the batch is nearly full, while stopping selection as soon as the batch reaches max_tis. The regression test verifies the batch is filled in two queries instead of four.

[avolant]

Tested it live in our Airflow deployed instances and got the following logs:

  Selected 1 of 32 task instances after iteration 1; retrying after excluding known blocked candidates.
  Selected 29 of 32 task instances after iteration 1; retrying after excluding known blocked candidates.
  Selected 31 of 32 task instances after iteration 1; retrying after excluding known blocked candidates. 

Stating that we reached this line

[avolant]

This has been in our production datacenter for the last 2 weeks, and we haven't experienced any regression so far. Still looking for a review

[avolant]

Hello @ashb / @XD-DENG would you have some time to review my PR? :)

[avolant]

Hi @potiuk / @hussein-awala 👋
Can I have a review on this PR please?

[potiuk]

Thanks for the detailed write-up and the production data. The diagnosis is right. On main, is_done = executable_tis or ... ends the selection loop as soon as a page yields a single executable TI, so a page dominated by concurrency-blocked higher-priority TIs queues one TI per scheduler loop. Re-querying with the learned starvation filters and excluding the already-selected IDs is the right fix, and so is moving executor_slots_available out of the per-page loop.

A few things before this can go in:

1. Rebase. #66878 moved the query into _build_schedulable_tis_query(...), so the selected-ID exclusion needs to become a parameter there, applied before the row_number() window as you have it now.
2. Bound the refill loop. Termination is guaranteed, but the worst-case number of iterations now grows with max_tis (up to roughly max_tis refills when each page yields one runnable TI). All of them run inside the critical section, holding the pool row locks and, on Postgres, the advisory lock, with NOT IN lists that keep growing. With max_tis_per_query=0 that can be hundreds of queries while other HA schedulers get critical_section_busy. Please add a small hard cap on refill iterations, and log at debug or emit a counter when it is hit.
3. Tests for the two new invariants. Neither is covered right now:
   • executor slots consumed across iterations, e.g. slots_available=2, page 1 = blocked + 1 runnable, page 2 = 2 runnable → exactly 2 queued. Moving the block back inside the loop currently passes the whole suite.
   • the in-page if len(executable_tis) >= max_tis: break: make page 2 hold more runnable TIs than the remaining capacity and assert the batch never exceeds max_tis. In the current test the batch fills on the last row of page 2, so deleting that break still passes.
4. Small: asserting session.scalars.call_count == 2 ties the test to unrelated internals. The queued-set assertion is the important one. Please also update the PR description, which still describes the LIMIT 1 refill.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

[avolant]

Thanks for the review. All four points are addressed:

1. Rebased onto main. The selected-ID exclusion is now the keyword-only excluded_ti_ids parameter of _build_schedulable_tis_query, applied before the row_number() window together with the starvation filters.
2. Refill queries are capped at MAX_TI_REFILL_QUERIES_PER_LOOP = 3, an internal constant next to MAX_PARTITION_DAG_RUNS_PER_LOOP. Hitting the cap logs at debug level and increments scheduler.critical_section_refill_limit_reached, which is added to the metrics registry. The production logs earlier in this thread only show retries after the first query, so three leaves headroom. The cap counts only queries issued after the first TI is selected. Queries before that stay uncapped as on main, because every loop restarts with empty filters, and a cap there could starve runnable TIs indefinitely behind enough blocked ones.
3. Added one parametrized test over a scenario where each three-row page holds a single runnable TI until the last page:
   • executor_slots_shared_across_queries uses two executor slots with max_tis=3 and expects exactly two queued TIs. It fails when the slot map is rebuilt per query.
   • fills_batch_up_to_max_tis ends on a page with two runnable TIs for one free position and expects exactly three. It fails without the in-page break.

A separate test patches the cap to 1, expects the batch to stop after one refill, and checks the counter.
4. Dropped the session.scalars.call_count assertion and updated the description.

One more behaviour change surfaced while writing the tests. main also ends the loop on a short page, but a run's max_active_tasks window can return only its blocked TIs while hiding its runnable ones. The PR keeps querying in that case, and test_select_task_instances_to_queue_finds_runnable_hidden_by_max_active_tasks_window covers it.

[avolant]

The test [Tests (AMD) / Low dep tests:core / core:LowestDeps:14:3.10:Core...Serialization (pull_request)](https://github.com/apache/airflow/actions/runs/37334353594/job/111852913998?pr=70307)Failing after 28m failure is unrelated to my changes. Maybe caused by #74035?

[avolant]

Thanks @hussein-awala for your review. You were indeed correct, and I updated my PR to account with this specificity. I added your test, reproduced the failure before my changes, and my new changes now make the test passes.

[hussein-awala]

Thanks for addressing the review feedback. The run-scoped exclusions fix the priority issue I raised, and the refill cap and regression tests cover the other concerns. LGTM, pending green CI.

One performance consideration: on main, the scheduler can already keep querying without a cap until it finds a runnable task, provided each page is full. This PR also allows that scanning to continue after short pages when new filters can expose runnable candidates.
That can increase critical-section duration for blocked-heavy workloads, although refills after the first selection are capped at three. This isn’t a blocker for me.

@ashb @uranusjr could you also take a look, particularly at the refill behavior and its impact on the scheduler’s critical section?

[Copilot]

🔵 Needs a closer look

Refills extend the scheduler’s lock-holding critical section, warranting human review and PostgreSQL validation after the rebase.

0 open findings

What changed in this PR

Improves scheduler throughput by refilling task batches when concurrency-blocked candidates leave capacity unused.

Changes:

• Continues selection while preserving executor capacity and excluding already-selected tasks.
• Scopes starvation filters to individual Dag runs and caps refill queries with a new metric.
• Adds regression tests for batch limits, priority ordering, and blocked candidates.

File	Description
shared/​observability/​src/​airflow_shared/​observability/​metrics/​metrics_template.yaml	Registers the refill-limit counter.
airflow-core/​tests/​unit/​jobs/​test_scheduler_job.py	Tests refill behavior and scheduling constraints.
airflow-core/​src/​airflow/​jobs/​scheduler_job_runner.py	Implements bounded refills and Dag-run-specific exclusions.

🧠 Review effort: Balanced

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

Handled.

[hussein-awala]

All the comments were addressed, I will merge. Thanks @avolant for the fix!