hansjone/netx
1
0
Fork 0
mirror of https://github.com/hansjone/netx.git synced 2026-10-11 12:53:17 +08:00
Code Issues Projects Releases 1 Packages Wiki Activity Actions

Make topology resume kick sync immediately and keep schedule across restarts.

Browse source 
Wire topology_auto_sync into resume/pause force-sync hints, ignore stale_running cleanup rows when computing the interval clock, and wait a full interval instead of pulling topology on every cold start.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-08-06 18:51:44 +08:00
parent 4fcec36514
commit 4bb1df8299
2 changed files with 30 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

35
netx_api/ume_support.py
View file

@ -358,20 +358,29 @@ def _sleep_or_until_paused(task_id: str, total_s: float) -> None:
def _last_finished_job_ended_at(db: Session, domain: str) -> datetime | None: def _last_finished_job_ended_at(db: Session, domain: str) -> datetime | None:
"""Latest finished sync job end time for domain (success or failed).""" """Latest finished sync job end time for domain (success or failed).
row = (
Ignores startup/reaper cleanup rows (``stale_running_*``) so process restart
does not reset the schedule clock to "just now".
"""
rows = (
db.query(UmeSyncJob) db.query(UmeSyncJob)
.filter( .filter(
UmeSyncJob.domain == domain, UmeSyncJob.domain == domain,
UmeSyncJob.ended_at.isnot(None), UmeSyncJob.ended_at.isnot(None),
) )
.order_by(UmeSyncJob.ended_at.desc()) .order_by(UmeSyncJob.ended_at.desc())
.limit(1) .limit(50)
.first() .all()
) )
if not row or row.ended_at is None: for row in rows:
return None err = str(getattr(row, "error_message", "") or "")
return _ensure_utc(row.ended_at) if "stale_running" in err:
continue
if row.ended_at is None:
continue
return _ensure_utc(row.ended_at)
return None
def _seconds_since_last_finished_job(db: Session, domain: str) -> float | None: def _seconds_since_last_finished_job(db: Session, domain: str) -> float | None:
@ -419,6 +428,18 @@ def _maybe_wait_for_sync_interval(
db.close() db.close()
_refresh_runtime_task_idle(task_id, domain) _refresh_runtime_task_idle(task_id, domain)
if elapsed is None: 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.
if domain == "topology":
_schedule_log.info(
"%s: no prior finished job for %s, wait full %ss (no startup pull)",
label,
domain,
interval_s,
)
_sleep_or_until_paused(task_id, float(interval_s))
return
_schedule_log.info("%s: no prior finished job for %s, sync now", label, domain) _schedule_log.info("%s: no prior finished job for %s, sync now", label, domain)
return return
if elapsed >= float(interval_s): if elapsed >= float(interval_s):

4
netx_api/ume_sync_router.py
View file

@ -221,7 +221,7 @@ def ume_runtime_task_pause(task: str) -> dict[str, Any]:
if tid not in UME_KNOWN_RUNTIME_TASKS: if tid not in UME_KNOWN_RUNTIME_TASKS:
raise HTTPException(status_code=404, detail="unknown_runtime_task") raise HTTPException(status_code=404, detail="unknown_runtime_task")
_runtime_pause_task(tid) _runtime_pause_task(tid)
if tid in ("alarms_current_auto_sync", "inventory_auto_sync"): if tid in ("alarms_current_auto_sync", "inventory_auto_sync", "topology_auto_sync"):
_clear_force_resume_hints(tid) _clear_force_resume_hints(tid)
if tid == "alarms_current_ws_consumer": if tid == "alarms_current_ws_consumer":
request_ws_reconnect() request_ws_reconnect()
@ -237,7 +237,7 @@ def ume_runtime_task_resume(task: str) -> dict[str, Any]:
if tid not in UME_KNOWN_RUNTIME_TASKS: if tid not in UME_KNOWN_RUNTIME_TASKS:
raise HTTPException(status_code=404, detail="unknown_runtime_task") raise HTTPException(status_code=404, detail="unknown_runtime_task")
_runtime_resume_task(tid) _runtime_resume_task(tid)
if tid in ("alarms_current_auto_sync", "inventory_auto_sync"): if tid in ("alarms_current_auto_sync", "inventory_auto_sync", "topology_auto_sync"):
_request_force_sync_after_resume(tid) _request_force_sync_after_resume(tid)
resume_hint = RT_RESUMED_SYNC_SOON resume_hint = RT_RESUMED_SYNC_SOON
elif tid == "alarms_current_ws_consumer": elif tid == "alarms_current_ws_consumer":