0821

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():
+        if product == "fox100":
-            nonlocal host_suffix, bell_prefix, netbox_time_ms
-
             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
+