refactore worker manager

app/core/event_hub.py

@@ -73,25 +73,25 @@ class EventHub:
 
     # app/core/event_hub.py
 
-def emit_analytics_trigger(self, org_id: str, source_id: str, extra: dict | None = None):
-    """Write trigger to single stream (worker manager polls this)"""
-    stream_key = "stream:analytics_triggers"
+    def emit_analytics_trigger(self, org_id: str, source_id: str, extra: dict | None = None):
+        """Write trigger to single stream (worker manager polls this)"""
+        stream_key = "stream:analytics_triggers"
     
-    payload = {
-        "org_id": org_id,
-        "source_id": source_id,
-        "timestamp": datetime.utcnow().isoformat(),
-    }
-    if extra:
-        payload.update(extra)
-
-    try:
-        msg_id = self.redis.xadd(stream_key, {"message": json.dumps(payload)})
-        logger.info(f"[hub] 📤 emitted trigger to {stream_key} (id: {msg_id})")
-        return msg_id
-    except Exception as e:
-        logger.error(f"[hub] ❌ failed to emit trigger: {e}", exc_info=True)
-        return None
+        payload = {
+            "org_id": org_id,
+            "source_id": source_id,
+            "timestamp": datetime.utcnow().isoformat(),
+        }
+        if extra:
+            payload.update(extra)
+
+        try:
+            msg_id = self.redis.xadd(stream_key, {"message": json.dumps(payload)})
+            logger.info(f"[hub] 📤 emitted trigger to {stream_key} (id: {msg_id})")
+            return msg_id
+        except Exception as e:
+            logger.error(f"[hub] ❌ failed to emit trigger: {e}", exc_info=True)
+            return None
 
     def ensure_consumer_group(self, stream_key: str, group: str):
         try:

app/core/worker_manager.py

@@ -4,31 +4,46 @@ import asyncio
 import json
 from typing import Dict
 from app.tasks.analytics_worker import AnalyticsWorker
-from app.redis_client import redis  # Use your existing Upstash client
+from app.redis_client import redis
 import logging
 
 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"
-        self.poll_interval = 1.0  # Seconds between polls
+        
+        # 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
+        
+        # Track idle state
+        self.consecutive_empty_polls = 0
     
     async def start_listener(self):
-        """🎧 Poll Upstash HTTP API for trigger events"""
-        logger.info(f"[manager] 🎧 listening via HTTP polling (interval: {self.poll_interval}s)")
+        """🎧 Adaptive polling: fast when busy, slow when idle"""
+        logger.info(
+            f"[manager] 🎧 listening (active: {self.poll_interval}s, idle: {self.idle_poll_interval}s)"
+        )
         
         while True:
             try:
-                # CRITICAL: Remove 'block' parameter - not supported by Upstash
+                # Poll Redis stream (non-blocking)
                 result = redis.xread({self.stream_key: self.last_id}, count=10)
                 
-                if result:
-                    # Parse messages
-                    messages = result.get(self.stream_key, [])
+                # Check if we got messages
+                has_messages = bool(result and isinstance(result, dict) and result.get(self.stream_key))
+                
+                if has_messages:
+                    # Reset idle counter
+                    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:
@@ -40,22 +55,40 @@ class WorkerManager:
                             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  # Update resume point
+                                self.last_id = msg_id
                             else:
-                                logger.warning(f"[manager] ⚠️ missing IDs in payload: {payload}")
+                                logger.warning(f"[manager] ⚠️ missing IDs: {payload}")
+                        
                         except json.JSONDecodeError as e:
                             logger.error(f"[manager] ❌ JSON error: {e}")
+                        except Exception as e:
+                            logger.error(f"[manager] ❌ message processing error: {e}", exc_info=True)
                     
-                    # Small delay after processing batch
+                    # Brief pause after batch, then continue polling fast
                     await asyncio.sleep(0.1)
+                    continue  # Skip to next iteration (stay in active mode)
+                
+                # No messages found
+                self.consecutive_empty_polls += 1
+                
+                # Choose sleep duration based on idle state
+                if self.consecutive_empty_polls > self.idle_threshold:
+                    delay = self.idle_poll_interval
+                    logger.debug(
+                        f"[manager] 💤 idle mode, sleeping {delay}s (empty polls: {self.consecutive_empty_polls})"
+                    )
                 else:
-                    # CRITICAL: Sleep when no messages
-                    logger.debug("[manager] 💤 no messages, sleeping")
-                    await asyncio.sleep(self.poll_interval)
-                    
+                    delay = self.poll_interval
+                    logger.debug(
+                        f"[manager] 💤 active mode, sleeping {delay}s (empty polls: {self.consecutive_empty_polls})"
+                    )
+                
+                await asyncio.sleep(delay)
+            
             except Exception as e:
                 logger.error(f"[manager] ❌ polling error: {e}", exc_info=True)
                 await asyncio.sleep(5)  # Back off on error
+                self.consecutive_empty_polls = 0  # Reset after error
     
     async def spawn_worker(self, org_id: str, source_id: str):
         """Spawn worker with distributed lock"""
@@ -63,7 +96,7 @@ class WorkerManager:
         lock_key = f"worker_lock:{worker_key}"
         
         try:
-            # Check existing lock
+            # Check if worker is already running
             if redis.exists(lock_key):
                 logger.debug(f"[manager] ⏭️ worker locked: {worker_key}")
                 return
@@ -72,7 +105,7 @@ class WorkerManager:
             redis.setex(lock_key, self.lock_ttl, "1")
             logger.info(f"[manager] 🔒 lock acquired: {lock_key}")
             
-            # Create worker
+            # Create and track worker
             worker = AnalyticsWorker(org_id, source_id)
             self.active_workers[worker_key] = worker
             
@@ -84,16 +117,17 @@ class WorkerManager:
             redis.delete(lock_key)
     
     async def _run_worker(self, worker: AnalyticsWorker, worker_key: str, lock_key: str):
-        """Execute worker and cleanup"""
+        """Run worker and cleanup"""
         try:
             results = await worker.run()
             logger.info(f"[manager] ✅ worker complete: {worker_key}")
         except Exception as e:
             logger.error(f"[manager] ❌ worker failed: {worker_key} - {e}", exc_info=True)
         finally:
-            # Cleanup
+            # Cleanup worker and lock
             self.active_workers.pop(worker_key, None)
             redis.delete(lock_key)
-            logger.info(f"[manager] 🧹 cleaned up: {worker_key}")
+            logger.debug(f"[manager] 🧹 cleaned up: {worker_key}")
+
 
 worker_manager = WorkerManager()

app/main.py

@@ -3,7 +3,7 @@
 MutSyncHub Analytics Engine
 Enterprise-grade AI analytics platform with zero-cost inference
 # """
-# import logging
+import logging
 import os
 import time
 import uuid
@@ -86,54 +86,49 @@ async def lifespan(app: FastAPI):
     except Exception as e:
         logger.error(f"🔴 Startup health check failed: {e}")
     
-    # Start scheduler in background
-    scheduler_process = subprocess.Popen(["python", "/app/scheduler_loop.py"])
-    logger.info(f"✅ Scheduler started (PID: {scheduler_process.pid})")
+    # Start scheduler in background (optional - controllable via env)
+    scheduler_process = None
+    if os.getenv("DISABLE_SCHEDULER") != "1":
+        try:
+            scheduler_process = subprocess.Popen(["python", "/app/scheduler_loop.py"])
+            logger.info(f"✅ Scheduler started (PID: {scheduler_process.pid})")
+        except Exception as e:
+            logger.warning(f"⚠️ Scheduler failed to start: {e}")
+    else:
+        logger.info("ℹ️ Scheduler start skipped (DISABLE_SCHEDULER=1)")
     
     logger.info("✅ Startup sequence complete")
 
-    # Setup Redis streams
-    logger.info("🔄 Setting up Redis streams...")
-    try:
-        active_orgs = event_hub.keys("entity:*")
-        for key in active_orgs:
-            key_parts = safe_redis_decode(key).split(":")
-            if len(key_parts) >= 3:
-                org_id, source_id = key_parts[1], key_parts[2]
-                stream_key = f"stream:analytics:{org_id}:{source_id}"
-                try:
-                    event_hub.ensure_consumer_group(stream_key, f"analytics_consumers_{org_id}")
-                except Exception as e:
-                    if "BUSYGROUP" not in str(e):
-                        logger.warning(f"⚠️ Stream setup warning: {e}")
-
-        logger.info("✅ Redis streams consumer groups ready")
-    except Exception as e:
-        logger.error(f"❌ Stream setup failed: {e}")
     
-    # ✅ start event hub (ensures consumer groups)
-    logger.info("🔌 initializing event hub...")
-    try:
-        # Ensure a sample consumer group exists (no-op if already present)
-        event_hub.ensure_consumer_group("stream:analytics:org_synth_123:default", "analytics_consumers")
-    except Exception as e:
-        logger.warning(f"Event hub init warning: {e}")
+   
+    # ✅ start worker manager listener (optional)
+    if os.getenv("DISABLE_WORKER_MANAGER") != "1":
+        logger.info("🚀 starting worker manager...")
+        try:
+            asyncio.create_task(worker_manager.start_listener(), name="worker-manager")
+        except Exception as e:
+            logger.error(f"❌ Failed to start worker manager: {e}")
+    else:
+        logger.info("ℹ️ Worker manager start skipped (DISABLE_WORKER_MANAGER=1)")
+    # Now load optional services (LLM, QStash)
+    if os.getenv("DISABLE_LLM_LOAD") != "1":
+        try:
+            load_llm_service()  # Starts background loading
+            logger.info("🤖 LLM service loading in background...")
+        except Exception as e:
+            logger.error(f"❌ LLM load failed: {e}")
+    else:
+        logger.info("ℹ️ LLM loading skipped (DISABLE_LLM_LOAD=1)")
 
-    # ✅ start worker manager listener
-    logger.info("🚀 starting worker manager...")
-    asyncio.create_task(worker_manager.start_listener(), name="worker-manager")
-    # Now load LLM service - it will use persistent cache
-    try:
-        load_llm_service()  # Starts background loading
-        logger.info("🤖 LLM service loading in background...")
-    except Exception as e:
-        logger.error(f"❌ LLM load failed: {e}")
-    # Continue anyway - LLM is optional for some features
-    try:
-        get_qstash_client()  # This creates the singleton if not exists
-        logger.info("✅ QStash ready")
-    except RuntimeError as e:
-        logger.warning(f"⚠️ QStash disabled: {e}")
+    # QStash client is optional; guard behind env var
+    if os.getenv("DISABLE_QSTASH") != "1":
+        try:
+            get_qstash_client()  # This creates the singleton if not exists
+            logger.info("✅ QStash ready")
+        except RuntimeError as e:
+            logger.warning(f"⚠️ QStash disabled: {e}")
+    else:
+        logger.info("ℹ️ QStash initialization skipped (DISABLE_QSTASH=1)")
     yield
     
     # ─── Shutdown ──────────────────────────────────────────────────────────────