fix(proxy): shift staggered job fire times on UTC instants

This commit is contained in:
Prasanna Jain 2026-10-24 22:30:40 -04:00
parent de5dc5b238
commit bca683cde3
2 changed files with 52 additions and 4 deletions

View file

@ -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:

View file

@ -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",
[