From 94e55887a7d2ee282f729951baf89b99ab3bad49 Mon Sep 17 00:00:00 2001 From: pavel Date: Sun, 15 Feb 2026 13:29:01 +0200 Subject: [PATCH] 1329 --- files/nats_registration_listener.py | 29 +++++++++-------------------- 1 file changed, 9 insertions(+), 20 deletions(-) diff --git a/files/nats_registration_listener.py b/files/nats_registration_listener.py index ec93de5..f8cd0b2 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 @@ -58,12 +58,6 @@ RMQ_TIMEOUT = 3.0 # Example: 600000 = 10 minutes RMQ_DELAY_MS = 15000 -# ---- Per-task delay overrides (human-editable) -# When action_next equals this task name, special delay rules apply. -POSTURE_SCAN_SOT = "sot-updater-scheduler" -# Delay to use when re-scheduling POSTURE_SCAN_SOT after a posture scan already happened. -POSTURE_SOT_WAITTIME_MS = 30000 - # Product -> Tag slug mapping (kept unchanged, though not used now) PRODUCT_TAG_SLUG = { "fox100": "fox100-auto-upgrade-latest", @@ -473,11 +467,6 @@ async def main(): else: # Publish task_name=action_next effective_delay_ms = RMQ_DELAY_MS - if action_next_str == POSTURE_SCAN_SOT and action_last_str == action_next_str: - effective_delay_ms = 0 - elif action_next_str == POSTURE_SCAN_SOT and action_last_str != action_next_str and action_last_str not in (None, ""): - effective_delay_ms = 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" @@ -518,10 +507,13 @@ async def main(): except Exception: routed = False + # ---- SURGICAL FIX ---- + # For delayed publishes (effective_delay_ms > 0), routed may be false but the message is accepted. publish_ok = True - if target_exchange != RMQ_EXCHANGE_DELAYED and not routed: + if effective_delay_ms <= 0 and not routed: publish_ok = False await log_status(f"[{ts()}] rmq: publish immediate routed=false host={host}") + # ---------------------- if publish_ok: # On success: set action_last and action_next_timestamp and action_state @@ -545,13 +537,8 @@ async def main(): bell_prefix = "\a" * 3 else: - # action_last != action_next -> publish immediately + # action_last != action_next -> publish effective_delay_ms = RMQ_DELAY_MS - if action_next_str == POSTURE_SCAN_SOT and action_last_str == action_next_str: - effective_delay_ms = 0 - elif action_next_str == POSTURE_SCAN_SOT and action_last_str != action_next_str and action_last_str not in (None, ""): - effective_delay_ms = 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" @@ -592,10 +579,12 @@ async def main(): except Exception: routed = False + # ---- SURGICAL FIX ---- publish_ok = True - if target_exchange != RMQ_EXCHANGE_DELAYED and not routed: + if effective_delay_ms <= 0 and not routed: publish_ok = False await log_status(f"[{ts()}] rmq: publish immediate routed=false host={host}") + # ---------------------- if publish_ok: now_epoch = int(time.time())