From 3088136aae729f19eb3a52684dba37c57a304124 Mon Sep 17 00:00:00 2001 From: oliver Date: Thu, 6 Aug 2026 18:54:38 +0800 Subject: [PATCH] Fix topology resume so it actually interrupts wait and starts a sync. Stop clearing the wake event at sleep entry, keep force-sync skip until consumed, and make pause loops wait on the resume event so UI start kicks topology immediately. Co-authored-by: Cursor --- netx_api/ume_runtime.py | 6 ++--- netx_api/ume_support.py | 59 +++++++++++++++++++++++++++++++---------- 2 files changed, 48 insertions(+), 17 deletions(-) diff --git a/netx_api/ume_runtime.py b/netx_api/ume_runtime.py index e6bed21..2e80a83 100644 --- a/netx_api/ume_runtime.py +++ b/netx_api/ume_runtime.py @@ -137,7 +137,7 @@ def start_api_sideband_threads() -> None: ume_support._runtime_is_paused("alarms_current_auto_sync"), ) if ume_support._runtime_is_paused("alarms_current_auto_sync"): - time.sleep(1) + ume_support._debounce_wake_event("alarms_current_auto_sync").wait(timeout=1.0) continue if is_startup_alarm_sync_pending(): ume_support._refresh_runtime_task_idle( @@ -235,7 +235,7 @@ def start_api_sideband_threads() -> None: ume_support._runtime_is_paused("inventory_auto_sync"), ) if ume_support._runtime_is_paused("inventory_auto_sync"): - time.sleep(1) + ume_support._debounce_wake_event("inventory_auto_sync").wait(timeout=1.0) continue ume_support._maybe_wait_for_sync_interval( task_id="inventory_auto_sync", @@ -304,7 +304,7 @@ def start_api_sideband_threads() -> None: ume_support._runtime_is_paused("topology_auto_sync"), ) if ume_support._runtime_is_paused("topology_auto_sync"): - time.sleep(1) + ume_support._debounce_wake_event("topology_auto_sync").wait(timeout=1.0) continue ume_support._maybe_wait_for_sync_interval( task_id="topology_auto_sync", diff --git a/netx_api/ume_support.py b/netx_api/ume_support.py index aa2114f..8aae8f6 100644 --- a/netx_api/ume_support.py +++ b/netx_api/ume_support.py @@ -334,14 +334,39 @@ def _run_startup_alarm_sync_before_ws() -> None: complete_startup_alarm_sync_gate() -def _sleep_or_until_paused(task_id: str, total_s: float) -> None: - """Sleep up to total_s wall seconds; honor pause; wake early on resume (debounce interrupt).""" +def _consume_force_sync_skip(task_id: str) -> bool: + """Return True once if resume/kick asked to skip the next debounce wait.""" + with _UME_DEBOUNCE_MUTEX: + if task_id in _UME_SYNC_SKIP_DEBOUNCE: + _UME_SYNC_SKIP_DEBOUNCE.discard(task_id) + return True + return False + + +def _has_force_sync_skip(task_id: str) -> bool: + with _UME_DEBOUNCE_MUTEX: + return task_id in _UME_SYNC_SKIP_DEBOUNCE + + +def _sleep_or_until_paused(task_id: str, total_s: float) -> bool: + """Sleep up to total_s; honor pause; wake early on resume. + + Returns True when the wait was interrupted for a forced sync (resume/kick). + """ + if _has_force_sync_skip(task_id): + return True deadline = time.time() + max(0.0, float(total_s)) ev = _debounce_wake_event(task_id) - ev.clear() + # Do not clear a wake that resume already set between skip-check and sleep. while time.time() < deadline: + if _has_force_sync_skip(task_id): + return True if _runtime_is_paused(task_id): - time.sleep(1) + # Pause must still be interruptible so resume is not stuck up to 1s+interval. + if ev.wait(timeout=1.0): + ev.clear() + if _has_force_sync_skip(task_id) or not _runtime_is_paused(task_id): + return True continue remaining = deadline - time.time() if remaining <= 0: @@ -350,11 +375,11 @@ def _sleep_or_until_paused(task_id: str, total_s: float) -> None: if ev.wait(timeout=timeout): ev.clear() _schedule_log.info("%s: debounce wait interrupted (resume)", task_id) - with _UME_DEBOUNCE_MUTEX: - _UME_SYNC_SKIP_DEBOUNCE.discard(task_id) - return + return True if ev.is_set(): ev.clear() + return _has_force_sync_skip(task_id) + return False def _last_finished_job_ended_at(db: Session, domain: str) -> datetime | None: @@ -416,17 +441,19 @@ def _maybe_wait_for_sync_interval( label: str, ) -> None: """Sleep until interval elapsed since last finished job (ended_at), if any.""" - with _UME_DEBOUNCE_MUTEX: - if task_id in _UME_SYNC_SKIP_DEBOUNCE: - _UME_SYNC_SKIP_DEBOUNCE.discard(task_id) - _schedule_log.info("%s: debounce skipped (resume/kick)", label) - return + if _consume_force_sync_skip(task_id): + _schedule_log.info("%s: debounce skipped (resume/kick)", label) + return db = SessionLocal() try: elapsed = _seconds_since_last_finished_job(db, domain) 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) + 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 @@ -438,7 +465,9 @@ def _maybe_wait_for_sync_interval( domain, interval_s, ) - _sleep_or_until_paused(task_id, float(interval_s)) + if _sleep_or_until_paused(task_id, float(interval_s)): + _consume_force_sync_skip(task_id) + _schedule_log.info("%s: wait interrupted — sync now", label) return _schedule_log.info("%s: no prior finished job for %s, sync now", label, domain) return @@ -447,7 +476,9 @@ def _maybe_wait_for_sync_interval( return wait_s = float(interval_s) - elapsed _schedule_log.info("%s: last finished %.0fs ago, wait %.0fs before sync", label, elapsed, wait_s) - _sleep_or_until_paused(task_id, 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) def _parse_time(text: str | None) -> datetime | None: