diff --git a/litellm/proxy/common_utils/scheduled_job_stagger.py b/litellm/proxy/common_utils/scheduled_job_stagger.py index e48e9686f13..7f2816fd3cb 100644 --- a/litellm/proxy/common_utils/scheduled_job_stagger.py +++ b/litellm/proxy/common_utils/scheduled_job_stagger.py @@ -33,7 +33,7 @@ import hashlib import os import socket from collections.abc import Callable, Mapping, Sequence -from datetime import datetime, timedelta +from datetime import datetime, timedelta, timezone from types import MappingProxyType from typing import Final, Protocol @@ -125,9 +125,9 @@ class _OffsetTrigger: self.offset = offset def get_next_fire_time(self, previous_fire_time: datetime | None, now: datetime) -> datetime | None: - shifted_previous: Final = None if previous_fire_time is None else previous_fire_time - self.offset - next_fire_time: Final = self.base.get_next_fire_time(shifted_previous, now - self.offset) - return None if next_fire_time is None else next_fire_time + self.offset + shifted_previous: Final = None if previous_fire_time is None else _shift(previous_fire_time, -self.offset) + next_fire_time: Final = self.base.get_next_fire_time(shifted_previous, _shift(now, -self.offset)) + return None if next_fire_time is None else _shift(next_fire_time, self.offset) def __str__(self) -> str: return f"{self.base}[+{int(self.offset.total_seconds())}s]" @@ -137,6 +137,17 @@ class _OffsetTrigger: BaseTrigger.register(_OffsetTrigger) +def _shift(moment: datetime, delta: timedelta) -> datetime: + """ + ``moment`` moved by ``delta`` of elapsed time, back in its own zone. + + Plain ``moment + delta`` is wall-clock arithmetic that drops ``fold``: inside a DST fall-back + hour it lands an hour early, the next fire never gets past ``now``, and APScheduler keeps + appending missed run times until the process runs out of memory. + """ + return (moment.astimezone(timezone.utc) + delta).astimezone(moment.tzinfo) + + def parse_stagger_settings(general_settings: Mapping[str, object]) -> ScheduledJobStaggerSettings: raw: Final = general_settings.get(GENERAL_SETTINGS_KEY) if raw is None: diff --git a/tests/unit/proxy/common_utils/test_scheduled_job_stagger.py b/tests/unit/proxy/common_utils/test_scheduled_job_stagger.py index 3eabfc5c840..e180cf57224 100644 --- a/tests/unit/proxy/common_utils/test_scheduled_job_stagger.py +++ b/tests/unit/proxy/common_utils/test_scheduled_job_stagger.py @@ -3,6 +3,7 @@ import logging import os from datetime import datetime, timedelta, timezone from types import SimpleNamespace +from zoneinfo import ZoneInfo import pytest from apscheduler.executors.asyncio import AsyncIOExecutor @@ -22,6 +23,7 @@ from litellm.proxy.common_utils.scheduled_job_stagger import ( stagger_trigger, ) +PARIS = ZoneInfo("Europe/Paris") OPERATOR_CRON_JOB_ID = "spend_log_cleanup_job" SHARED_INTERVAL_JOB_IDS = ("periodic_reload_job", "add_deployment_job") @@ -250,6 +252,41 @@ def test_a_job_registered_after_startup_still_gets_its_offset(): assert _fire_times(shifted, start, 2)[1] - _fire_times(shifted, start, 1)[0] == timedelta(seconds=3600) +@pytest.mark.parametrize( + "start", + [ + datetime(2026, 10, 25, 2, 30, fold=1, tzinfo=PARIS), # second pass of the fall-back hour + datetime(2026, 3, 29, 1, 58, tzinfo=PARIS), # just before the spring-forward gap + ], + ids=["fall-back", "spring-forward"], +) +@pytest.mark.parametrize( + "base", + [ + IntervalTrigger(seconds=60, start_date=datetime(2026, 1, 1, tzinfo=PARIS), timezone=PARIS), + CronTrigger(minute="*", timezone=PARIS), + ], + ids=["interval", "cron"], +) +def test_a_staggered_job_keeps_firing_once_a_minute_across_a_dst_change(base, start): + """ + Inside the repeated fall-back hour the shifted fire used to land an hour before ``now``, + so APScheduler never caught up and grew its list of missed run times without bound + """ + shifted = stagger_trigger( + job_id="periodic_reload_job", + trigger=base, + period_seconds=60, + settings=_settings(offsets={"periodic_reload_job": 30}), + identity="pod-a:1", + ) + + fires = _fire_times(shifted, start, 4) + + assert 0 < fires[0].timestamp() - start.timestamp() <= 60 + assert [later.timestamp() - earlier.timestamp() for earlier, later in itertools.pairwise(fires)] == [60] * 3 + + @pytest.mark.parametrize( "raw, expected_window", [