diff --git a/litellm/proxy/common_utils/scheduled_job_stagger.py b/litellm/proxy/common_utils/scheduled_job_stagger.py index e48e9686f13..1fb06e37167 100644 --- a/litellm/proxy/common_utils/scheduled_job_stagger.py +++ b/litellm/proxy/common_utils/scheduled_job_stagger.py @@ -33,9 +33,9 @@ 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 +from typing import Final, Protocol, overload from apscheduler.events import EVENT_JOB_SUBMITTED from apscheduler.triggers.base import BaseTrigger @@ -108,6 +108,19 @@ class JobSubmission(Protocol): def scheduled_run_times(self) -> Sequence[datetime]: ... +@overload +def _shift(dt: None, delta: timedelta) -> None: ... +@overload +def _shift(dt: datetime, delta: timedelta) -> datetime: ... +def _shift(dt: datetime | None, delta: timedelta) -> datetime | None: + if dt is None: + return None + if dt.tzinfo is not None: + # Perform arithmetic in UTC to preserve monotonic timeline across DST transitions (#48286) + return (dt.astimezone(timezone.utc) + delta).astimezone(dt.tzinfo) + return dt + delta + + class _OffsetTrigger: """ Delegates to ``base`` on a clock rolled back by ``offset``, then rolls the answer @@ -125,9 +138,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 = _shift(previous_fire_time, -self.offset) + next_fire_time: Final = self.base.get_next_fire_time(shifted_previous, _shift(now, -self.offset)) + return _shift(next_fire_time, self.offset) def __str__(self) -> str: return f"{self.base}[+{int(self.offset.total_seconds())}s]" diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 6fcf789dc1b..ba02778ff7d 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -1916,7 +1916,7 @@ async def _proxy_lifespan(app: FastAPI, resources: AsyncExitStack) -> AsyncGener "disable_model_info_refresh" in general_settings and general_settings["disable_model_info_refresh"] is True ) model_info_scheduler: Final = ( - None if model_info_refresh_disabled else scheduler if scheduler is not None else AsyncIOScheduler() + None if model_info_refresh_disabled else scheduler if scheduler is not None else AsyncIOScheduler(timezone="UTC") ) if model_info_scheduler is not None: model_info_scheduler.add_job( @@ -11587,8 +11587,8 @@ class ProxyStartupEvent: executors={ "default": scheduler_executor, }, - # Disable timezone awareness to reduce computation - timezone=None, + # Explicitly use UTC timezone to prevent infinite loops and unbounded memory growth during DST transitions (#48285) + timezone="UTC", ) # Use fixed intervals with small random offset instead of jitter 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..7351b33ceea 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 @@ -14,6 +15,7 @@ from apscheduler.triggers.interval import IntervalTrigger from litellm.constants import PTU_ROLLUP_JOB_ID, PTU_ROLLUP_LOCK_TTL_SECONDS from litellm.proxy._types import ScheduledJobStaggerSettings from litellm.proxy.common_utils.scheduled_job_stagger import ( + _OffsetTrigger, apply_scheduled_job_stagger, attach_job_timing_logger, offset_seconds, @@ -309,3 +311,35 @@ def test_job_timing_is_logged_with_scheduled_and_actual_start(caplog): assert f"scheduled_run_time={scheduled.isoformat()}" in message assert "actual_start_time=" in message assert "delay=2." in message + + +def test_offset_trigger_dst_fallback_does_not_loop(): + """ + Regression test for #48285 / #48286. + During DST fall-back (repeated hour, fold=1), naive timedelta arithmetic drops fold + and sends shifted_previous 1 hour back in time, causing Job._get_run_times to spin + infinitely and exhaust memory. UTC-safe shifting ensures monotonic progress. + """ + tz = ZoneInfo("Europe/Paris") + base = IntervalTrigger(seconds=60, timezone=tz) + staggered = _OffsetTrigger(base, timedelta(seconds=30)) + + def count_advances(trigger, now: datetime, cap: int = 100) -> int: + nrt = datetime.fromtimestamp(now.timestamp() - 90, tz) + n = 0 + while nrt and nrt.astimezone(timezone.utc) <= now.astimezone(timezone.utc): + n += 1 + if n >= cap: + return cap + nrt = trigger.get_next_fire_time(nrt, now) + return n + + for label, now in [ + ("01:30 CEST", datetime(2026, 10, 25, 1, 30, tzinfo=tz)), + ("02:30 CET (repeated hour)", datetime(2026, 10, 25, 2, 30, fold=1, tzinfo=tz)), + ("03:30 CET", datetime(2026, 10, 25, 3, 30, tzinfo=tz)), + ]: + plain_count = count_advances(base, now) + staggered_count = count_advances(staggered, now) + assert staggered_count == plain_count == 2, f"Failed for {label}: staggered={staggered_count}, plain={plain_count}" +