pavel/nats-scheduler
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1329

Browse Source
This commit is contained in:
pavel
2026-02-15 13:29:01 +02:00
parent c268337621
commit 94e55887a7
1 changed files with 9 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
files/nats_registration_listener.py
View File

@@ -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())