fix: pace failed calendar refresh retries
This commit is contained in:
@@ -10,7 +10,7 @@ from sqlalchemy import or_, select
|
|||||||
from sqlalchemy.ext.asyncio import AsyncSession
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||||||
|
|
||||||
from . import calendar as calendar_service
|
from . import calendar as calendar_service
|
||||||
from .models import CalendarSubscription, utcnow
|
from .models import CalendarSubscription
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
DEFAULT_REFRESH_INTERVAL = timedelta(minutes=15)
|
DEFAULT_REFRESH_INTERVAL = timedelta(minutes=15)
|
||||||
@@ -22,7 +22,9 @@ async def refresh_subscription_cache(
|
|||||||
row: CalendarSubscription,
|
row: CalendarSubscription,
|
||||||
*,
|
*,
|
||||||
fail_without_cache: bool = True,
|
fail_without_cache: bool = True,
|
||||||
|
attempted_at: datetime | None = None,
|
||||||
) -> bool:
|
) -> bool:
|
||||||
|
attempt_time = attempted_at or datetime.now(UTC)
|
||||||
requested_url = row.url
|
requested_url = row.url
|
||||||
requested_version = getattr(row, "updated_at", None)
|
requested_version = getattr(row, "updated_at", None)
|
||||||
try:
|
try:
|
||||||
@@ -51,7 +53,8 @@ async def refresh_subscription_cache(
|
|||||||
row.ics_cache = result.content.decode("utf-8-sig")
|
row.ics_cache = result.content.decode("utf-8-sig")
|
||||||
row.etag = result.etag
|
row.etag = result.etag
|
||||||
row.last_modified = result.last_modified
|
row.last_modified = result.last_modified
|
||||||
row.refreshed_at = utcnow()
|
row.refreshed_at = attempt_time
|
||||||
|
row.updated_at = attempt_time
|
||||||
row.last_error = None
|
row.last_error = None
|
||||||
await db.commit()
|
await db.commit()
|
||||||
return True
|
return True
|
||||||
@@ -61,6 +64,7 @@ async def refresh_subscription_cache(
|
|||||||
await db.rollback()
|
await db.rollback()
|
||||||
raise HTTPException(502, error) from exc
|
raise HTTPException(502, error) from exc
|
||||||
row.last_error = error
|
row.last_error = error
|
||||||
|
row.updated_at = attempt_time
|
||||||
await db.commit()
|
await db.commit()
|
||||||
return False
|
return False
|
||||||
|
|
||||||
@@ -76,13 +80,15 @@ async def refresh_due_subscriptions(
|
|||||||
rows = (await db.scalars(select(CalendarSubscription).where(
|
rows = (await db.scalars(select(CalendarSubscription).where(
|
||||||
CalendarSubscription.enabled.is_(True),
|
CalendarSubscription.enabled.is_(True),
|
||||||
or_(
|
or_(
|
||||||
CalendarSubscription.refreshed_at.is_(None),
|
CalendarSubscription.updated_at.is_(None),
|
||||||
CalendarSubscription.refreshed_at <= cutoff,
|
CalendarSubscription.updated_at <= cutoff,
|
||||||
),
|
),
|
||||||
).order_by(CalendarSubscription.refreshed_at, CalendarSubscription.created_at))).all()
|
).order_by(CalendarSubscription.refreshed_at, CalendarSubscription.created_at))).all()
|
||||||
for row in rows:
|
for row in rows:
|
||||||
try:
|
try:
|
||||||
await refresh_subscription_cache(db, row, fail_without_cache=False)
|
await refresh_subscription_cache(
|
||||||
|
db, row, fail_without_cache=False, attempted_at=current
|
||||||
|
)
|
||||||
except Exception:
|
except Exception:
|
||||||
await db.rollback()
|
await db.rollback()
|
||||||
logger.exception("Unexpected calendar refresh failure", extra={"subscription_id": str(row.id)})
|
logger.exception("Unexpected calendar refresh failure", extra={"subscription_id": str(row.id)})
|
||||||
|
|||||||
@@ -351,3 +351,58 @@ def test_refresh_discards_response_when_url_changes_in_flight(monkeypatch):
|
|||||||
assert row.url == "https://example.com/new.ics"
|
assert row.url == "https://example.com/new.ics"
|
||||||
assert row.ics_cache is None
|
assert row.ics_cache is None
|
||||||
assert db.committed is False
|
assert db.committed is False
|
||||||
|
|
||||||
|
|
||||||
|
def test_failed_source_waits_until_next_interval_and_does_not_abort_batch(client, monkeypatch):
|
||||||
|
client = initialized(client)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
"backend.calendar.validate_calendar_url",
|
||||||
|
lambda url: (url, "93.184.216.34", 443),
|
||||||
|
)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
"backend.calendar.fetch_calendar",
|
||||||
|
lambda *args, **kwargs: FetchResult(ICS, None, None, False),
|
||||||
|
)
|
||||||
|
first = client.post(
|
||||||
|
"/api/v1/calendar-subscriptions",
|
||||||
|
json={"name": "first", "url": "https://example.com/first.ics"},
|
||||||
|
).json()
|
||||||
|
client.post(
|
||||||
|
"/api/v1/calendar-subscriptions",
|
||||||
|
json={"name": "second", "url": "https://example.com/second.ics"},
|
||||||
|
)
|
||||||
|
attempts = []
|
||||||
|
|
||||||
|
def fetch(url, **kwargs):
|
||||||
|
attempts.append(url)
|
||||||
|
if url.endswith("first.ics"):
|
||||||
|
raise HTTPException(502, "upstream down")
|
||||||
|
return FetchResult(ICS, None, None, False)
|
||||||
|
|
||||||
|
monkeypatch.setattr("backend.calendar.fetch_calendar", fetch)
|
||||||
|
|
||||||
|
async def refresh_twice():
|
||||||
|
from sqlalchemy.ext.asyncio import async_sessionmaker
|
||||||
|
|
||||||
|
from backend.db import get_engine
|
||||||
|
|
||||||
|
factory = async_sessionmaker(get_engine(), expire_on_commit=False)
|
||||||
|
now = datetime.now(UTC) + timedelta(minutes=16)
|
||||||
|
async with factory() as db:
|
||||||
|
first_count = await refresh_due_subscriptions(
|
||||||
|
db, now=now, refresh_interval=timedelta(minutes=15)
|
||||||
|
)
|
||||||
|
async with factory() as db:
|
||||||
|
second_count = await refresh_due_subscriptions(
|
||||||
|
db, now=now + timedelta(minutes=1), refresh_interval=timedelta(minutes=15)
|
||||||
|
)
|
||||||
|
return first_count, second_count
|
||||||
|
|
||||||
|
assert asyncio.run(refresh_twice()) == (2, 0)
|
||||||
|
assert attempts == [
|
||||||
|
"https://example.com/first.ics",
|
||||||
|
"https://example.com/second.ics",
|
||||||
|
]
|
||||||
|
listed = client.get("/api/v1/calendar-subscriptions").json()
|
||||||
|
failed = next(item for item in listed if item["id"] == first["id"])
|
||||||
|
assert failed["stale"] is True
|
||||||
|
|||||||
Reference in New Issue
Block a user