From d3c3a32e7c6bbac8715eb39a32d541b014886be2 Mon Sep 17 00:00:00 2001 From: pavel Date: Sun, 15 Feb 2026 09:57:13 +0200 Subject: [PATCH] 0957 --- files/nats_registration_listener.py | 46 +++++++++++++++++------------ 1 file changed, 27 insertions(+), 19 deletions(-) diff --git a/files/nats_registration_listener.py b/files/nats_registration_listener.py index 866e04b..247a186 100644 --- a/files/nats_registration_listener.py +++ b/files/nats_registration_listener.py @@ -58,6 +58,10 @@ RMQ_TIMEOUT = 3.0 # Example: 600000 = 10 minutes RMQ_DELAY_MS = 15000 +# ---- Posture scan special-case (user-requested variables) +POSTURE_SCAN_SOT = "sot-updater-wrapper" # "posture-scan-sot" +POSTURE_SOT_WAITTIME_MS = 30000 # "posture-sot-waittime" (ms) + # Product -> Tag slug mapping (kept unchanged, though not used now) PRODUCT_TAG_SLUG = { "fox100": "fox100-auto-upgrade-latest", @@ -424,6 +428,7 @@ 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: @@ -464,8 +469,15 @@ async def main(): flush=True, ) else: - # Publish task_name=action_next - target_exchange = RMQ_EXCHANGE_DELAYED if RMQ_DELAY_MS > 0 else RMQ_EXCHANGE_WORK + # ---- Delay override logic (user-requested) + effective_delay_ms = RMQ_DELAY_MS + if action_next_str == POSTURE_SCAN_SOT: + if action_last_str == action_next_str: + effective_delay_ms = 0 + else: + effective_delay_ms = int(POSTURE_SOT_WAITTIME_MS) + + target_exchange = RMQ_EXCHANGE_DELAYED if effective_delay_ms > 0 else RMQ_EXCHANGE_WORK rmq_url = f"http://{RMQ_HOST}:{RMQ_PORT}/api/exchanges/{RMQ_VHOST}/{target_exchange}/publish" payload_obj = { @@ -484,8 +496,8 @@ async def main(): "payload_encoding": "string", } - if RMQ_DELAY_MS > 0: - rmq_body["properties"]["headers"] = {"x-delay": int(RMQ_DELAY_MS)} + if effective_delay_ms > 0: + rmq_body["properties"]["headers"] = {"x-delay": int(effective_delay_ms)} await log_status( f"[{ts()}] ok, here i will execute\n" @@ -504,11 +516,7 @@ async def main(): routed = bool((resp or {}).get("routed", False)) except Exception: routed = False - - # FIX: delayed exchange accepts messages even if routed=false - publish_ok = True if target_exchange == RMQ_EXCHANGE_DELAYED else routed - - if not publish_ok: + if not routed: 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 @@ -538,8 +546,13 @@ async def main(): bell_prefix = "\a" * 3 else: - # action_last != action_next -> publish immediately - target_exchange = RMQ_EXCHANGE_DELAYED if RMQ_DELAY_MS > 0 else RMQ_EXCHANGE_WORK + # action_last != action_next -> publish + # ---- Delay override logic (user-requested) + effective_delay_ms = RMQ_DELAY_MS + if action_next_str == POSTURE_SCAN_SOT: + effective_delay_ms = int(POSTURE_SOT_WAITTIME_MS) + + target_exchange = RMQ_EXCHANGE_DELAYED if effective_delay_ms > 0 else RMQ_EXCHANGE_WORK rmq_url = f"http://{RMQ_HOST}:{RMQ_PORT}/api/exchanges/{RMQ_VHOST}/{target_exchange}/publish" payload_obj = { @@ -558,8 +571,8 @@ async def main(): "payload_encoding": "string", } - if RMQ_DELAY_MS > 0: - rmq_body["properties"]["headers"] = {"x-delay": int(RMQ_DELAY_MS)} + if effective_delay_ms > 0: + rmq_body["properties"]["headers"] = {"x-delay": int(effective_delay_ms)} await log_status( f"[{ts()}] ok, here i will execute\n" @@ -578,11 +591,7 @@ async def main(): routed = bool((resp or {}).get("routed", False)) except Exception: routed = False - - # FIX: delayed exchange accepts messages even if routed=false - publish_ok = True if target_exchange == RMQ_EXCHANGE_DELAYED else routed - - if not publish_ok: + if not routed: await log_status(f"[{ts()}] rmq: publish immediate routed=false host={host}") else: now_epoch = int(time.time()) @@ -663,4 +672,3 @@ if __name__ == "__main__": asyncio.run(main()) except KeyboardInterrupt: pass -