GitHub user VectorPeak added a comment to the discussion: Tasks with a newer logical_date are never scheduled to run when max_active_tis_per_dag is used
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](https://apache.googlesource.com/airflow/%2B/refs/tags/providers-common-compat/1.6.0b1/airflow/jobs/scheduler_job_runner.py)) 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: ```python 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](https://airflow.apache.org/docs/apache-airflow/2.10.4/administration-and-deployment/priority-weight.html?utm_source=openai)) Roughly: ```python 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 😆 GitHub link: https://github.com/apache/airflow/discussions/69010#discussioncomment-17554991 ---- This is an automatically sent email for [email protected]. To unsubscribe, please send an email to: [email protected]
