From 655813d4d97fccc3ac26bc04eac9e46ab41f3c8b Mon Sep 17 00:00:00 2001 From: eezd Date: Thu, 5 Nov 2026 13:32:19 +0800 Subject: [PATCH] Preserve calendar failure cooldown across worker restarts --- backend/src/alphapulse/calendar/service.py | 2 +- backend/src/alphapulse/calendar/worker.py | 10 +++++++++- backend/tests/test_calendar_recovery.py | 19 +++++++++++++++++++ 3 files changed, 29 insertions(+), 2 deletions(-) diff --git a/backend/src/alphapulse/calendar/service.py b/backend/src/alphapulse/calendar/service.py index 83034fa..c401939 100644 --- a/backend/src/alphapulse/calendar/service.py +++ b/backend/src/alphapulse/calendar/service.py @@ -93,7 +93,7 @@ async def save_schedule(session: AsyncSession, source: str, url: str, events: li row.stage = event.stage if row.release_url is None and event.release_url is not None: row.release_url = event.release_url - row.next_check_at = max(now, event.at) + row.next_check_at = max(row.next_check_at, now, event.at) if getattr(row, "collection_failures", 0) else max(now, event.at) row.revision_no += 1 record_revision(session, row, raw_id) # 仅撤回尚未公布的未来事件；过去记录不因官网滚动窗口而被删除。 diff --git a/backend/src/alphapulse/calendar/worker.py b/backend/src/alphapulse/calendar/worker.py index ab8d780..a09457d 100644 --- a/backend/src/alphapulse/calendar/worker.py +++ b/backend/src/alphapulse/calendar/worker.py @@ -178,13 +178,17 @@ class CalendarWorker: urls += [("eia", eia.SCHEDULE)] urls += [("adp", adp.INDEX)] async with SessionLocal() as session: - disabled = set(await session.scalars(select(Source.code).where(Source.code.in_(BOUNDED_SOURCES), Source.enabled.is_(False)))) + controls = list(await session.scalars(select(Source).where(Source.code.in_(BOUNDED_SOURCES)))) + disabled = {source.code for source in controls if not source.enabled} + paused = {source.code for source in controls if source.paused_until and source.paused_until > now} urls = [(s, u) for s, u in urls if s not in disabled] checked = {} # 启动时明确各来源尚未就绪；月份滚动后移除不再检查的旧月状态。 self.sources = {url: self.sources.get(url, {"source": source, "checked_at": now.isoformat(), "error": "尚未同步"}) for source, url in urls} for source, url in urls: + if source in paused: + continue retry_at = self.sources[url].get("retry_at") if retry_at and datetime.fromisoformat(retry_at) > datetime.now(UTC): continue @@ -263,9 +267,13 @@ class CalendarWorker: source.consecutive_failures = 0 source.last_success_at = datetime.now(UTC) source.last_error = None + source.paused_until = None else: source.consecutive_failures += 1 source.last_error = outcomes[0] + deadlines = [datetime.fromisoformat(state['retry_at']) for state in self.sources.values() + if state['source'] == code and state.get('retry_at')] + source.paused_until = max([datetime.now(UTC) + timedelta(hours=1), *deadlines]) if source.consecutive_failures >= 3: source.enabled = False await session.commit() diff --git a/backend/tests/test_calendar_recovery.py b/backend/tests/test_calendar_recovery.py index c05d179..50ce1f8 100644 --- a/backend/tests/test_calendar_recovery.py +++ b/backend/tests/test_calendar_recovery.py @@ -249,3 +249,22 @@ def test_dol_lagging_index_uses_only_expected_official_archive(trading): await worker.http.aclose() await worker.bls.aclose() asyncio.run(run()) + + +def test_restart_honors_persisted_source_pause(trading, monkeypatch): + from alphapulse.calendar import worker as module + session = AsyncMock() + session.__aenter__.return_value = session + session.scalars.return_value = [SimpleNamespace(code='adp', enabled=True, paused_until=datetime.now(UTC) + timedelta(hours=1))] + session.get.return_value = None + monkeypatch.setattr(module, 'SessionLocal', lambda: session) + async def run(): + worker = module.CalendarWorker() + worker.fetch = AsyncMock(side_effect=httpx.ConnectError('offline')) + try: + await worker.schedules() + assert all(call.args[0] != 'adp' for call in worker.fetch.await_args_list) + finally: + await worker.http.aclose() + await worker.bls.aclose() + asyncio.run(run())