diff --git a/files/nats_registration_listener.py b/files/nats_registration_listener.py index 9a4b656..d75178f 100644 --- a/files/nats_registration_listener.py +++ b/files/nats_registration_listener.py @@ -1,12 +1,13 @@ #!/usr/bin/env python3 """ -NATS Registration Listener (compact one-line output) ---------------------------------------------------- +NATS Registration Listener (compact one-line output + strict read-only dedupe) +------------------------------------------------------------------------------ Print exactly one line per message: [TS] product= mac= fw= -- Keeps your existing CLI flags (servers/subject/queue/auth/TLS). -- Writes are serialized to avoid garbled output. +- Listener-only (does NOT publish/reply). +- Writes are serialized to avoid interleaved output. +- Strict dedupe by exact payload bytes within a short TTL (env DEDUP_TTL, default 2.0s). - Falls back gracefully if fields/structure are missing. Install deps: @@ -20,6 +21,8 @@ import os import signal import ssl import sys +import hashlib +from time import monotonic from datetime import datetime, timezone from typing import Optional, Dict, Any @@ -140,6 +143,11 @@ async def main(): async with print_lock: print(s, file=sys.stderr, flush=True) + # --- Strict read-only dedupe config (env) --- + # Drop exact duplicate payloads seen within this TTL window. + DEDUPE_TTL = float(os.environ.get("DEDUP_TTL", "2.0")) + recent_payloads: Dict[bytes, float] = {} # digest -> expires_at (monotonic seconds) + reconnect_time_wait = 2 max_reconnect_attempts = -1 @@ -173,20 +181,37 @@ async def main(): ) async def message_handler(msg: nats.aio.msg.Msg): - line = None - now = ts() + now_s = ts() payload = msg.data - # Best-effort JSON parse + # --- STRICT DEDUPE by exact payload bytes (read-only) --- + # Use a stable, compact digest to key the recent map. + digest = hashlib.blake2b(payload, digest_size=16).digest() + nowm = monotonic() + exp = recent_payloads.get(digest) + if exp and exp > nowm: + return # drop exact duplicate seen very recently + recent_payloads[digest] = nowm + DEDUPE_TTL + + # Optional light cleanup to keep the dict bounded + if len(recent_payloads) > 4096: + # remove expired entries + cutoff = nowm + recent_payloads_keys = list(recent_payloads.keys()) + for k in recent_payloads_keys: + if recent_payloads.get(k, 0) <= cutoff: + recent_payloads.pop(k, None) + + # Parse and print one line + product = mac = fw = "-" try: text = payload.decode("utf-8", errors="replace") obj = json.loads(text) product, mac, fw = extract_fields(obj) - line = f"[{now}] product={product} mac={mac} fw={fw}" except Exception: - # If not JSON or broken, still emit a line that shows size only - line = f"[{now}] product=- mac=- fw=-" + pass + line = f"[{now_s}] product={product} mac={mac} fw={fw}" if args.include_subject: line += f" subject={msg.subject}" @@ -199,13 +224,15 @@ async def main(): else: await nc.subscribe(args.subject, cb=message_handler) - await log_status(f"[{ts()}] Listening on subject '{args.subject}' (queue={getattr(args, 'queue', None) or '-'}) via {args.servers}") + await log_status( + f"[{ts()}] Listening on subject '{args.subject}' (queue={getattr(args, 'queue', None) or '-'}) via {args.servers} " + f"| mode=one-line dedupe=payload ttl={DEDUPE_TTL}s tx=disabled" + ) # Graceful shutdown stop_event = asyncio.Event() def handle_signal(*_): - # Use stderr for status asyncio.create_task(log_status(f"[{ts()}] Received stop signal, draining...")) stop_event.set()