Tasks with a newer logical_date are never scheduled to run when max_active_tis_per_dag is used

[franzwong]

Due to issues with the external data provider, newer data may be available before older data. The order of the data is not a concern for me, so I allow the tasks to keep retrying throughout the day. I have limited the number of task instances. However, tasks from newer DAG runs are never executed because Airflow always prioritizes tasks from older DAG runs.

I created a simple example to reproduce this. There are 6 active DAG runs. Airflow only executes tasks from the first 4 DAG runs and never executes tasks from the remaining 2 DAG runs. I tried using a pool with 2 slots and 6 DAGs, but the result is the same.

Any thoughts on this? Do I need to implement a custom scheduler? Thanks.

import pendulum
import time

from airflow import DAG
from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk.exceptions import AirflowException


def example(**context):
    print("Before sleep")
    time.sleep(30)
    print("After sleep")
    raise AirflowException("Testing")


with DAG(
    dag_id="example_dag",
    start_date=pendulum.datetime(2026, 5, 1, 0, 0, tz="UTC"),
    schedule="* * * * *",
    catchup=False,
    max_active_runs=6,
    is_paused_upon_creation=False,
    tags=["example"],
) as dag:
    PythonOperator(
        task_id="example",
        python_callable=example,
        retries=300,
        retry_delay=pendulum.duration(seconds=30),
        max_active_tis_per_dag=2,
    )

[VectorPeak]

This looks like the scheduler behaving consistently with the current policy, not the pool behaving oddly. In the critical scheduling path, Airflow picks scheduled TIs ordered by priority weight descending, then DagRun.logical_date ascending, then map_index; after that it applies constraints such as pool slots, DAG/task concurrency, and max_active_tis_per_dag. With one task and equal priority, the oldest eligible run always wins the tie. A pool with 2 slots changes the capacity, not the ordering, so it does not make the selection fair across logical dates. (apache.googlesource.com)

The retry pattern in the repro makes the starvation much easier to hit. The first two runs execute, fail, and enter retry delay. While they are waiting, the next two can run. When the first two become eligible again, they are still older than runs 5 and 6, so they get picked again. That creates a loop where the first few active runs keep cycling and the newest active runs never get a chance.

A custom scheduler is probably more machinery than you want here. If missing data is an expected condition, I would avoid representing “data is not ready yet” as AirflowException + hundreds of retries on the processing task. That keeps producing normal task failures and keeps the same old logical dates at the front of the scheduler’s candidate set.

A cleaner shape is usually:

• one cheap “is this partition ready” step, preferably a deferrable sensor or mode="reschedule" sensor
• one actual processing task, limited by a pool or max_active_tis_per_dag
• make unavailable partitions wait without consuming worker/pool capacity
• only let the processing task run once the provider says that partition is available

For example, conceptually:

wait_for_data >> process_data

Put the pool / task concurrency limit on process_data, not on a task that both waits and processes. If there is no built-in sensor for your provider, a small custom sensor is much less invasive than a custom scheduler.

If you really want to keep this retry-based design and only change ordering, look at priority weights before touching the scheduler. Since priority is sorted before logical_date, a custom PriorityWeightStrategy can make repeatedly failing old runs lose priority, or make newer logical dates more important. Airflow supports custom weight rules for this kind of scheduling policy, and the docs even show a strategy that decreases priority by retry attempt. (airflow.apache.org)

Roughly:

class NewerOrFewerRetriesFirst(PriorityWeightStrategy):
 def get_weight(self, ti):
 # Example only: tune this policy for your case.
 return max(1000 - ti.try_number, 1)

Then use it as the task’s weight_rule. That would not be a perfect fairness guarantee, but it is the intended extension point for changing scheduler preference among otherwise-runnable task instances.

If this is intended as a bug report rather than a design question, the useful next details would be the exact Airflow version, executor, database backend, and a short scheduler log excerpt showing the runs cycling. My read is that this is more of a scheduler fairness/improvement case than a pool or max_active_tis_per_dag bug, because the current ordering explicitly prefers older logical_date when priorities are equal.

---

If my answer solved your problem, you can click answered the question. I'm really here to help, and along the way I'm also collecting Galaxy Brain badges haha 😆

[franzwong]

Thanks. I will think about the suggestions you give.