fixed reddis hammering

app/core/worker_manager.py

@@ -1,129 +1,140 @@
-# app/core/worker_manager.py
+# app/core/worker_manager.py – TRUE ASYNC VERSION
 
 import asyncio
 import json
-from typing import Dict
-from app.tasks.analytics_worker import AnalyticsWorker
-from app.redis_client import redis
+import time
+from typing import Dict, Optional
 import logging
 
+from app.core.event_hub import event_hub
+from app.tasks.analytics_worker import trigger_kpi_computation
+
 logger = logging.getLogger(__name__)
 
 
 class WorkerManager:
     def __init__(self):
-        self.active_workers: Dict[str, AnalyticsWorker] = {}
-        self.lock_ttl = 300
-        self.stream_key = "stream:analytics_triggers"
-        self.last_id = "0"
-        
-        # Adaptive polling intervals
-        self.poll_interval = 1.0  # Active: 1s when messages are flowing
-        self.idle_poll_interval = 60  # Idle: 60s when no work
-        self.idle_threshold = 5  # 5 empty polls = idle mode
+        self.active_workers: Dict[str, asyncio.Task] = {}
+        self._shutdown = False
         
-        # Track idle state
-        self.consecutive_empty_polls = 0
+        # Smart polling intervals
+        self.stream_check_interval = 5.0  # Check stream every 5s max
+        self.cleanup_interval = 10.0      # Cleanup every 10s
+        self.last_stream_check = 0
+        self.last_cleanup = 0
     
     async def start_listener(self):
-        """🎧 Stream-based listener: block on Redis stream reads to avoid polling
-
-        NOTE: The original adaptive polling loop is left commented below for
-        reference. This implementation uses blocking XREAD (if supported) so
-        workers are only spawned when a readiness message is available.
         """
-        logger.info(f"[manager] 🎧 stream listener (blocking read) on {self.stream_key}")
-
-        # --- Original polling loop (commented) ---
-        # (kept for reference during refactor)
-        # logger.info(
-        #     f"[manager] 🎧 listening (active: {self.poll_interval}s, idle: {self.idle_poll_interval}s)"
-        # )
-        # while True:
-        #     ...
-
-        # New: blocking xread loop
-        while True:
-            try:
-                try:
-                    # Block for up to 5s waiting for new messages; fall back
-                    # to non-blocking if 'block' not supported by client
-                    result = redis.xread({self.stream_key: self.last_id}, count=10, block=5000)
-                except TypeError:
-                    # Client may not accept block kwarg; try without block
-                    result = redis.xread({self.stream_key: self.last_id}, count=10)
-
-                has_messages = bool(result and isinstance(result, dict) and result.get(self.stream_key))
-
-                if not has_messages:
-                    # No messages within block window; continue (loop will block again)
-                    continue
-
-                # We have messages — process them
-                self.consecutive_empty_polls = 0
-                messages = result[self.stream_key]
-                logger.info(f"[manager] 📥 received {len(messages)} messages")
-
-                for msg_id, msg_data in messages:
+        🎧 TRUE async listener: pubsub for triggers + throttled stream checks
+        Redis ops: ~0.2/sec idle, ~1-2/sec under load
+        """
+        logger.info("🎧 Worker Manager: Einstein+Elon mode ENGAGED (true async)")
+        
+        # Use pubsub for immediate triggers (ZERO polling)
+        pubsub = event_hub.redis.pubsub()
+        await pubsub.subscribe("analytics_trigger")
+        
+        try:
+            while not self._shutdown:
+                # 1. Non-blocking pubsub check (true push - zero Redis ops when idle)
+                message = await pubsub.get_message(timeout=0.1)
+                
+                if message and message["type"] == "message":
                     try:
-                        payload = json.loads(msg_data.get("message", "{}"))
-                        org_id = payload.get("org_id")
-                        source_id = payload.get("source_id")
-
-                        if org_id and source_id:
-                            logger.info(f"[manager] 🚀 processing {org_id}:{source_id}")
-                            await self.spawn_worker(org_id, source_id)
-                            self.last_id = msg_id
-                        else:
-                            logger.warning(f"[manager] ⚠️ missing IDs: {payload}")
-
-                    except json.JSONDecodeError as e:
-                        logger.error(f"[manager] ❌ JSON error: {e}")
+                        payload = json.loads(message["data"])
+                        await self._handle_trigger(payload)
                     except Exception as e:
-                        logger.error(f"[manager] ❌ message processing error: {e}", exc_info=True)
-
-            except Exception as e:
-                logger.error(f"[manager] ❌ streaming error: {e}", exc_info=True)
-                await asyncio.sleep(2)
+                        logger.error(f"[MANAGER] ❌ Trigger parse error: {e}")
+                
+                # 2. Throttled stream check (fallback safety net)
+                now = time.time()
+                if now - self.last_stream_check > self.stream_check_interval:
+                    await self._check_stream()
+                    self.last_stream_check = now
+                
+                # 3. Throttled cleanup
+                if now - self.last_cleanup > self.cleanup_interval:
+                    self._cleanup_completed_workers()
+                    self.last_cleanup = now
+                
+                # 4. Yield control (prevent CPU spinning)
+                await asyncio.sleep(0.01)
+                
+        except asyncio.CancelledError:
+            logger.info("[MANAGER] 🛑 Listener cancelled")
+        finally:
+            await pubsub.close()
+            logger.info("[MANAGER] 🔌 Pubsub closed")
     
-    async def spawn_worker(self, org_id: str, source_id: str):
-        """Spawn worker with distributed lock"""
-        worker_key = f"{org_id}:{source_id}"
-        lock_key = f"worker_lock:{worker_key}"
-        
+    async def _check_stream(self):
+        """Lightweight stream check - only reads new messages"""
         try:
-            # Check if worker is already running
-            if redis.exists(lock_key):
-                logger.debug(f"[manager] ⏭️ worker locked: {worker_key}")
+            # Non-blocking read of recent messages only
+            result = event_hub.redis.xrevrange(
+                "stream:analytics_triggers",
+                count=5  # Only check last 5
+            )
+            
+            if result:
+                logger.debug(f"[MANAGER] 📥 Stream check found {len(result)} messages")
+                for msg_id, data in result:
+                    payload = json.loads(data.get("message", "{}"))
+                    await self._handle_trigger(payload)
+                    # Acknowledge processed message
+                    event_hub.redis.xdel("stream:analytics_triggers", msg_id)
+                    
+        except Exception as e:
+            logger.debug(f"[MANAGER] Stream check failed: {e}")
+    
+    async def _handle_trigger(self, data: dict):
+        """Process trigger with rate limiting"""
+        try:
+            org_id = data.get("org_id")
+            source_id = data.get("source_id")
+            
+            if not org_id or not source_id:
+                logger.warning(f"[MANAGER] ⚠️ Invalid trigger: {data}")
                 return
             
-            # Set lock
-            redis.setex(lock_key, self.lock_ttl, "1")
-            logger.info(f"[manager] 🔒 lock acquired: {lock_key}")
+            worker_id = f"{org_id}:{source_id}"
             
-            # Create and track worker
-            worker = AnalyticsWorker(org_id, source_id)
-            self.active_workers[worker_key] = worker
+            # Skip if already running
+            if worker_id in self.active_workers and not self.active_workers[worker_id].done():
+                logger.debug(f"[MANAGER] ⏭️ Worker already active: {worker_id}")
+                return
             
-            # Run in background
-            asyncio.create_task(self._run_worker(worker, worker_key, lock_key))
+            # Spawn worker
+            logger.info(f"[MANAGER] 🚀 Spawning worker: {worker_id}")
+            task = asyncio.create_task(
+                self._run_worker(worker_id, org_id, source_id),
+                name=f"worker-{worker_id}"
+            )
+            self.active_workers[worker_id] = task
             
         except Exception as e:
-            logger.error(f"[manager] ❌ spawn failed: {e}", exc_info=True)
-            redis.delete(lock_key)
+            logger.error(f"[MANAGER] ❌ Trigger handling failed: {e}", exc_info=True)
     
-    async def _run_worker(self, worker: AnalyticsWorker, worker_key: str, lock_key: str):
-        """Run worker and cleanup"""
+    async def _run_worker(self, worker_id: str, org_id: str, source_id: str):
+        """Run the actual KPI computation"""
         try:
-            results = await worker.run()
-            logger.info(f"[manager] ✅ worker complete: {worker_key}")
+            # Use the existing trigger function (handles locks, caching, etc)
+            await trigger_kpi_computation(org_id, source_id)
+            logger.info(f"[MANAGER] ✅ Worker complete: {worker_id}")
         except Exception as e:
-            logger.error(f"[manager] ❌ worker failed: {worker_key} - {e}", exc_info=True)
+            logger.error(f"[MANAGER] ❌ Worker failed: {worker_id} - {e}", exc_info=True)
         finally:
-            # Cleanup worker and lock
-            self.active_workers.pop(worker_key, None)
-            redis.delete(lock_key)
-            logger.debug(f"[manager] 🧹 cleaned up: {worker_key}")
-
-
+            self.active_workers.pop(worker_id, None)
+    
+    def _cleanup_completed_workers(self):
+        """Remove completed tasks from registry"""
+        done_workers = [
+            wid for wid, task in self.active_workers.items()
+            if task.done()
+        ]
+        for wid in done_workers:
+            self.active_workers.pop(wid, None)
+            logger.debug(f"[MANAGER] 🧹 Cleaned up: {wid}")
+
+
+# Global instance
 worker_manager = WorkerManager()

app/main.py

@@ -14,7 +14,7 @@ import pathlib
 import json 
 
 # # ─── Third-Party ──────────────────────────────────────────────────────────────
-# from fastapi import FastAPI, Depends, HTTPException, Request, Query, BackgroundTasks
+from fastapi import FastAPI, Depends, HTTPException, Request, Query, BackgroundTasks
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 from contextlib import asynccontextmanager