From 6622d4ce7a9f6bcdefe53c1efdb2aa9cc7c8c7d0 Mon Sep 17 00:00:00 2001 From: oliver Date: Thu, 6 Aug 2026 18:59:52 +0800 Subject: [PATCH] Clarify topology schedule: next run is 24h after last real sync end. Resume still skips the wait and runs immediately; after that job writes ended_at, the following cycle waits a full interval from that completion (ignoring stale_running cleanup rows). Co-authored-by: Cursor --- netx_api/ume_support.py | 47 ++++++++++++++++++++++++++++------------- 1 file changed, 32 insertions(+), 15 deletions(-) diff --git a/netx_api/ume_support.py b/netx_api/ume_support.py index 8aae8f6..5aa2eae 100644 --- a/netx_api/ume_support.py +++ b/netx_api/ume_support.py @@ -383,16 +383,18 @@ def _sleep_or_until_paused(task_id: str, total_s: float) -> bool: def _last_finished_job_ended_at(db: Session, domain: str) -> datetime | None: - """Latest finished sync job end time for domain (success or failed). + """Anchor for the sync interval clock: last *real* job completion. - Ignores startup/reaper cleanup rows (``stale_running_*``) so process restart - does not reset the schedule clock to "just now". + Uses the latest ``ended_at`` among done/failed rows, skipping startup/reaper + cleanup (``stale_running_*``). So: resume → sync now → after that job + finishes, the next wait is a full interval from *that* ``ended_at``. """ rows = ( db.query(UmeSyncJob) .filter( UmeSyncJob.domain == domain, UmeSyncJob.ended_at.isnot(None), + UmeSyncJob.status.in_(("done", "failed")), ) .order_by(UmeSyncJob.ended_at.desc()) .limit(50) @@ -409,7 +411,7 @@ def _last_finished_job_ended_at(db: Session, domain: str) -> datetime | None: def _seconds_since_last_finished_job(db: Session, domain: str) -> float | None: - """Seconds since latest job with ended_at for domain (done or failed). None if none.""" + """Seconds since latest real finished job. None if none.""" end = _last_finished_job_ended_at(db, domain) if end is None: return None @@ -417,7 +419,7 @@ def _seconds_since_last_finished_job(db: Session, domain: str) -> float | None: def _refresh_runtime_task_idle(task_id: str, domain: str, *, last_error: str | None = None) -> None: - """Mark scheduled sync task running; last_run_at = last finished job time (idle / debounce).""" + """Mark scheduled sync task idle; last_run_at = last real finished job time.""" with _UME_RUNTIME_LOCK: prev_error = str((_UME_RUNTIME_TASKS.get(task_id) or {}).get("last_error") or "") db = SessionLocal() @@ -440,9 +442,16 @@ def _maybe_wait_for_sync_interval( interval_s: int, label: str, ) -> None: - """Sleep until interval elapsed since last finished job (ended_at), if any.""" + """Wait until ``interval_s`` since last real finished job, unless resume/kick. + + Flow for topology (and inventory): + - Resume/kick → skip wait → sync immediately. + - After that sync ends (job ``ended_at`` written) → next loop waits a full + interval from that completion. + - Process restart does not invent a new anchor from stale cleanup rows. + """ if _consume_force_sync_skip(task_id): - _schedule_log.info("%s: debounce skipped (resume/kick)", label) + _schedule_log.info("%s: debounce skipped (resume/kick) — sync now", label) return db = SessionLocal() try: @@ -450,17 +459,15 @@ def _maybe_wait_for_sync_interval( finally: db.close() _refresh_runtime_task_idle(task_id, domain) - # Resume may arrive while we were computing elapsed / refreshing idle status. if _consume_force_sync_skip(task_id): - _schedule_log.info("%s: debounce skipped after idle refresh (resume/kick)", label) + _schedule_log.info("%s: debounce skipped after idle refresh (resume/kick) — sync now", label) return if elapsed is None: - # Topology dumps are heavy: do not pull on every process start when there is - # no prior real finished job (e.g. only orphan cleanup rows). Wait one full - # interval unless resume/kick skipped debounce above. + # No real sync yet (only stale cleanup or empty history): wait one full + # interval so API restart does not pull topology immediately. if domain == "topology": _schedule_log.info( - "%s: no prior finished job for %s, wait full %ss (no startup pull)", + "%s: no prior real sync for %s, wait full %ss", label, domain, interval_s, @@ -472,10 +479,20 @@ def _maybe_wait_for_sync_interval( _schedule_log.info("%s: no prior finished job for %s, sync now", label, domain) return if elapsed >= float(interval_s): - _schedule_log.info("%s: last finished %.0fs ago (>= %ss), sync now", label, elapsed, interval_s) + _schedule_log.info( + "%s: last real sync %.0fs ago (>= %ss), sync now", + label, + elapsed, + interval_s, + ) return wait_s = float(interval_s) - elapsed - _schedule_log.info("%s: last finished %.0fs ago, wait %.0fs before sync", label, elapsed, wait_s) + _schedule_log.info( + "%s: last real sync %.0fs ago, wait %.0fs (next run = last ended_at + interval)", + label, + elapsed, + wait_s, + ) if _sleep_or_until_paused(task_id, wait_s): _consume_force_sync_skip(task_id) _schedule_log.info("%s: wait interrupted — sync now", label)