Fix scheduler to check per-schedule run counts for multi-schedule deployments
The scheduler's deployment selection query previously used aggregate
count/max_time across ALL schedules in the HAVING clause. This meant
that when a deployment had multiple schedules with different frequencies,
the high-frequency schedule could be starved of runs while the
low-frequency schedules kept the aggregate above the threshold.
Changed _get_select_deployments_to_schedule_query to use per-schedule
subqueries that check each active schedule's run count and max scheduled
time independently using idempotency_key prefix matching. A deployment
is now selected for re-scheduling if ANY of its active schedules has
fewer than min_runs future SCHEDULED runs or insufficient time coverage.
Added regression tests verifying:
- Query selects deployment when one schedule's runs are exhausted
- Scheduler generates new runs for starved schedules
Co-Authored-By: Nate Nowack <nate@prefect.io>
Keep latest timestamp in natural-key dedup within batch
Explicitly compare state_timestamp when deduplicating by natural key
instead of relying on dict-overwrite order, which could select an older
event when the same natural key appears across multiple task-run IDs.
Co-Authored-By: alex.s@prefect.io <ajstreed1@gmail.com>
Reject negative DateTimeDelta intervals in validation
Use in_months_days_secs_nanos() to check all components are
non-negative, preventing infinite loops in schedule generation
from negative intervals like DateTimeDelta(hours=-1).
Co-Authored-By: Nate Nowack <nate@prefect.io>