one liners 0623

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=<productName> mac=<mac> fw=<active>
 
-- Keeps your existing CLI flags (servers/subject/queue/auth/TLS).
+- Listener-only (does NOT publish/reply).
-- Writes are serialized to avoid garbled output.
+- 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_s = ts()
-        now = 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
+            pass
-            line = f"[{now}] product=- mac=- fw=-"
 
+        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()