From ef4820661c76fb54a97de05905441f9c54da66b4 Mon Sep 17 00:00:00 2001 From: pavel Date: Sun, 15 Feb 2026 08:21:09 +0200 Subject: [PATCH] 0821 --- files/nats_registration_listener.py | 58 ++++++++++++++--------------- 1 file changed, 28 insertions(+), 30 deletions(-) diff --git a/files/nats_registration_listener.py b/files/nats_registration_listener.py index e03ad4f..866e04b 100644 --- a/files/nats_registration_listener.py +++ b/files/nats_registration_listener.py @@ -49,7 +49,7 @@ RMQ_PORT = 15672 RMQ_USER = "admin" RMQ_PASS = "change_me" RMQ_VHOST = "app" -RMQ_EXCHANGE_WORK = "deviceconfig" # direct exchange (immediate) +RMQ_EXCHANGE_WORK = "deviceconfig" # direct exchange (immediate) RMQ_EXCHANGE_DELAYED = "deviceconfig.delayed" # delayed exchange (x-delayed-message) RMQ_ROUTING_KEY = "deviceconfig" RMQ_TIMEOUT = 3.0 @@ -316,18 +316,6 @@ async def main(): DEDUPE_TTL = float(os.environ.get("DEDUP_TTL", "2.0")) recent_payloads: Dict[bytes, float] = {} - # ---- Per-device lock keyed by normalized MAC (prevents overlap for same device) - mac_locks: Dict[str, asyncio.Lock] = {} - mac_locks_guard = asyncio.Lock() - - async def get_mac_lock(mac_key: str) -> asyncio.Lock: - async with mac_locks_guard: - lk = mac_locks.get(mac_key) - if lk is None: - lk = asyncio.Lock() - mac_locks[mac_key] = lk - return lk - async def disconnected_cb(): await log_status(f"[{ts()}] Disconnected from NATS.") @@ -387,9 +375,7 @@ async def main(): bell_prefix = "" # ASCII BEL when action_next present (3x) netbox_time_ms = 0.0 - async def handle_fox100(): - nonlocal host_suffix, bell_prefix, netbox_time_ms - + if product == "fox100": nb_start = time.perf_counter() try: host, iface_id, dev_id, status_val, tag_slugs, action_next, action_last, action_next_timestamp, action_state = nb_lookup_device_by_mac( @@ -401,7 +387,7 @@ async def main(): host_suffix += f" iface_id={iface_id}" if host: - # Gate on action_state: allow only "" or "ready" + # Gate on action_state: allow only "" or "ready" or "done" action_state_str = "" try: if action_state is None: @@ -438,7 +424,6 @@ async def main(): action_next_str = None if not has_action_next: - # User-requested behavior: if no action -> just shoot a message to stdout and we're ok async with print_lock: print(f"[{ts()}] no action_next for host={host}", file=sys.stdout, flush=True) else: @@ -519,7 +504,11 @@ async def main(): routed = bool((resp or {}).get("routed", False)) except Exception: routed = False - if not routed: + + # FIX: delayed exchange accepts messages even if routed=false + publish_ok = True if target_exchange == RMQ_EXCHANGE_DELAYED else routed + + if not publish_ok: await log_status(f"[{ts()}] rmq: publish immediate routed=false host={host}") else: # On success: set action_last and action_next_timestamp and action_state @@ -529,7 +518,13 @@ async def main(): "Accept": "application/json", "Authorization": f"Token {NB_TOKEN}", } - patch_body = {"custom_fields": {"action_last": action_next_str, "action_next_timestamp": str(now_epoch), "action_state": "started"}} + patch_body = { + "custom_fields": { + "action_last": action_next_str, + "action_next_timestamp": str(now_epoch), + "action_state": "started", + } + } _, pcode = http_patch_json( f"{base}/api/dcim/devices/{dev_id}/", patch_body, @@ -583,7 +578,11 @@ async def main(): routed = bool((resp or {}).get("routed", False)) except Exception: routed = False - if not routed: + + # FIX: delayed exchange accepts messages even if routed=false + publish_ok = True if target_exchange == RMQ_EXCHANGE_DELAYED else routed + + if not publish_ok: await log_status(f"[{ts()}] rmq: publish immediate routed=false host={host}") else: now_epoch = int(time.time()) @@ -593,7 +592,13 @@ async def main(): "Accept": "application/json", "Authorization": f"Token {NB_TOKEN}", } - patch_body = {"custom_fields": {"action_last": action_next_str, "action_next_timestamp": str(now_epoch), "action_state": "started"}} + patch_body = { + "custom_fields": { + "action_last": action_next_str, + "action_next_timestamp": str(now_epoch), + "action_state": "started", + } + } _, pcode = http_patch_json( f"{base}/api/dcim/devices/{dev_id}/", patch_body, @@ -611,14 +616,6 @@ async def main(): await nb_problem(log_status, f"[{ts()}] nb: unexpected error mac={mac!r} err={e!r}") netbox_time_ms = (time.perf_counter() - nb_start) * 1000 - if product == "fox100": - mac_key = normalize_mac(mac) - if mac_key: - async with (await get_mac_lock(mac_key)): - await handle_fox100() - else: - await handle_fox100() - total_ms = (monotonic() - t_start) * 1000 async with NB_PROBLEM_LOCK: @@ -666,3 +663,4 @@ if __name__ == "__main__": asyncio.run(main()) except KeyboardInterrupt: pass +