-check canonify triger timing-review worker startup and wait logic

.gitattributes

@@ -1 +1,5 @@
-*.duckdb filter=lfs diff=lfs merge=lfs -text
+# Do not LFS large runtime DBs; keep templates if needed
+*.duckdb -filter -merge -diff -text
+
+# If you want templates/fixtures to remain tracked, add an override
+# templates/*.duckdb filter=lfs diff=lfs merge=lfs -text

app/core/event_hub.py

@@ -19,7 +19,11 @@ class EventHub:
         return self.redis.get(key)
 
     def setex(self, key: str, ttl: int, value: str):
-        return self.redis.setex(key, ttl, value)
+        try:
+            return self.redis.setex(key, ttl, value)
+        except Exception as e:
+            logger.error(f"[hub] ❌ setex failed for {key}: {e}", exc_info=True)
+            raise
 
     def exists(self, key: str) -> bool:
         return self.redis.exists(key)
@@ -147,6 +151,17 @@ class EventHub:
     def keys(self, pattern: str):
         return self.redis.keys(pattern)
 
+    def pipeline(self):
+        """Return a redis pipeline-like object if supported by client.
+
+        Note: Upstash client may not support classic pipelines; callers should
+        handle attribute errors and fall back to sequential commands.
+        """
+        try:
+            return self.redis.pipeline()
+        except Exception:
+            return None
+
 
 # Singleton
 event_hub = EventHub()

app/core/worker_manager.py

@@ -26,69 +26,65 @@ class WorkerManager:
         self.consecutive_empty_polls = 0
     
     async def start_listener(self):
-        """🎧 Adaptive polling: fast when busy, slow when idle"""
-        logger.info(
-            f"[manager] 🎧 listening (active: {self.poll_interval}s, idle: {self.idle_poll_interval}s)"
-        )
-        
+        """🎧 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:
-                # Poll Redis stream (non-blocking)
-                result = redis.xread({self.stream_key: self.last_id}, count=10)
-                
-                # Check if we got messages
+                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 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:
-                        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}")
-                        except Exception as e:
-                            logger.error(f"[manager] ❌ message processing error: {e}", exc_info=True)
-                    
-                    # 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:
-                    delay = self.poll_interval
-                    logger.debug(
-                        f"[manager] 💤 active mode, sleeping {delay}s (empty polls: {self.consecutive_empty_polls})"
-                    )
-                
-                await asyncio.sleep(delay)
-            
+
+                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:
+                    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}")
+                    except Exception as e:
+                        logger.error(f"[manager] ❌ message processing error: {e}", exc_info=True)
+
             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
+                logger.error(f"[manager] ❌ streaming error: {e}", exc_info=True)
+                await asyncio.sleep(2)
     
     async def spawn_worker(self, org_id: str, source_id: str):
         """Spawn worker with distributed lock"""

app/mapper.py

@@ -119,59 +119,130 @@ def poll_for_entity(org_id: str, source_id: str, timeout: int = 30) -> dict:
     
     # 5. Emergency fallback (worker is dead)
     print("[poll] ⚠️ Both attempts failed - using direct detection")
-    entity_info = _fallback_detection(org_id, source_id)
-    
-    # 🎯 NEW: Force write to Redis (ensure it's there)
-    event_hub.setex(
-        f"entity:{org_id}:{source_id}", 
-        3600, 
-        json.dumps(entity_info)
-    )
-    
-    # 🎯 NEW: Clear stale cache so next read is fresh
-    if (org_id, source_id) in _ENTITY_CACHE:
-        del _ENTITY_CACHE[(org_id, source_id)]
-    
+
+    # Use the combined fallback so we only hit DuckDB once and write both
+    # entity AND industry keys atomically when possible.
+    entity_info, industry_info = _fallback_combined(org_id, source_id)
+
+    # Invalidate local cache entry so that subsequent callers read Redis first
+    _ENTITY_CACHE.pop((org_id, source_id), None)
+
     return entity_info
 
 
+# OLD: _fallback_detection kept for reference (commented out during refactor)
+"""
 def _fallback_detection(org_id: str, source_id: str) -> dict:
+    # (original implementation)
+    ...
+"""
+
+
+def _fallback_combined(org_id: str, source_id: str) -> tuple[dict, dict]:
+    """Single DuckDB query to produce both entity and industry detections.
+
+    Guarantees:
+    - Always writes `entity:{org_id}:{source_id}` and
+      `industry:{org_id}:{source_id}` to Redis (or logs if write fails).
+    - Updates module caches then invalidates them so readers re-check Redis.
+    - Attempts parallel detection to reduce latency.
     """
-    Emergency: Detect entity directly from DuckDB.
-    Writes result to Redis and cache for recovery.
-    """
-    print(f"[fallback] 🚨 Running fallback for {org_id}/{source_id}")
-    
-    conn = get_conn(org_id)
-    rows = conn.execute("""
+    print(f"[fallback_combined] 🚨 Running combined fallback for {org_id}/{source_id}")
+
+    # Default UNKNOWN placeholders
+    entity_info = {"entity_type": "UNKNOWN", "confidence": 0.0}
+    industry_info = {"industry": "UNKNOWN", "confidence": 0.0}
+
+    try:
+        conn = get_conn(org_id)
+        rows = conn.execute("""
             SELECT row_data
             FROM main.raw_rows
             WHERE row_data IS NOT NULL
             USING SAMPLE 100
         """).fetchall()
-    
-    if not rows:
-        print("[fallback] ❌ No data found, returning UNKNOWN")
-        entity_info = {"entity_type": "UNKNOWN", "confidence": 0.0}
-    else:
-        parsed = [json.loads(r[0]) for r in rows if r[0]]
-        df = pd.DataFrame(parsed)
-        
-        # This is fast (<50ms) with rule-based detection
-        entity_type, confidence, _ = hybrid_detect_entity_type(org_id, df, f"{source_id}.json")
-        entity_info = {"entity_type": entity_type, "confidence": confidence}
-        print(f"[fallback] ✅ Direct detection: {entity_type} ({confidence:.2%})")
-    
-    # ✅ CRITICAL: Write to Redis BEFORE returning
-    redis_key = f"entity:{org_id}:{source_id}"
-    event_hub.setex(redis_key, 3600, json.dumps(entity_info))
-    print(f"[fallback] 💾 WRITTEN TO REDIS: {redis_key}")
-    
-    # ✅ Also populate module cache for immediate reuse
-    cache_key = (org_id, source_id)
-    _ENTITY_CACHE[cache_key] = entity_info
-    
-    return entity_info
+
+        if rows:
+            parsed = [json.loads(r[0]) for r in rows if r[0]]
+            df = pd.DataFrame(parsed)
+            df.columns = [str(col).lower().strip() for col in df.columns]
+
+            # Run both detectors concurrently (thread pool for CPU/IO-bound work)
+            from concurrent.futures import ThreadPoolExecutor
+
+            def run_entity():
+                try:
+                    return hybrid_detect_entity_type(org_id, df, f"{source_id}.json")
+                except Exception as e:
+                    print(f"[fallback_combined] ❌ entity detection failed: {e}")
+                    return ("UNKNOWN", 0.0, False)
+
+            def run_industry():
+                try:
+                    from app.hybrid_industry_detector import hybrid_detect_industry_type
+                    return hybrid_detect_industry_type(org_id, df, source_id)
+                except Exception as e:
+                    print(f"[fallback_combined] ❌ industry detection failed: {e}")
+                    return ("UNKNOWN", 0.0, False)
+
+            with ThreadPoolExecutor(max_workers=2) as ex:
+                ent_f = ex.submit(run_entity)
+                ind_f = ex.submit(run_industry)
+                ent_res = ent_f.result()
+                ind_res = ind_f.result()
+
+            entity_info = {"entity_type": ent_res[0], "confidence": ent_res[1]}
+            industry_info = {"industry": ind_res[0], "confidence": ind_res[1]}
+
+            print(f"[fallback_combined] ✅ Entity: {entity_info['entity_type']} ({entity_info['confidence']:.2%})")
+            print(f"[fallback_combined] ✅ Industry: {industry_info['industry']} ({industry_info['confidence']:.2%})")
+
+    except Exception as e:
+        print(f"[fallback_combined] ❌ Combined fallback failed: {e}")
+
+    # Persist to Redis; prefer pipeline when available for minimal window
+    e_key = f"entity:{org_id}:{source_id}"
+    i_key = f"industry:{org_id}:{source_id}"
+
+    try:
+        pipe = event_hub.pipeline()
+        if pipe is not None:
+            try:
+                pipe.setex(e_key, 3600, json.dumps(entity_info))
+                pipe.setex(i_key, 3600, json.dumps(industry_info))
+                # Also add a per-source readiness nudging stream entry
+                try:
+                    # Structured readiness message for per-source stream
+                    msg = json.dumps({
+                        "org_id": org_id,
+                        "source_id": source_id,
+                        "status": "ready"
+                    })
+                    pipe.xadd(event_hub.stream_key(org_id, source_id), {"message": msg})
+                except Exception:
+                    # Not fatal; continue
+                    pass
+                pipe.execute()
+            except Exception as e:
+                print(f"[fallback_combined] ❌ Pipeline execute failed: {e}")
+                # Fall back to sequential writes
+                event_hub.setex(e_key, 3600, json.dumps(entity_info))
+                event_hub.setex(i_key, 3600, json.dumps(industry_info))
+        else:
+            # Pipeline not available; do sequential writes
+            event_hub.setex(e_key, 3600, json.dumps(entity_info))
+            event_hub.setex(i_key, 3600, json.dumps(industry_info))
+
+    except Exception as e:
+        print(f"[fallback_combined] ❌ Redis write failed: {e}")
+
+    # Update caches (then immediately invalidate to avoid stale-reads window)
+    _ENTITY_CACHE[(org_id, source_id)] = entity_info
+    _INDUSTRY_CACHE[(org_id, source_id)] = industry_info
+    _ENTITY_CACHE.pop((org_id, source_id), None)
+    _INDUSTRY_CACHE.pop((org_id, source_id), None)
+
+    return entity_info, industry_info
     #poll for industry from redis
 def poll_for_industry(org_id: str, source_id: str, timeout: int = 10) -> dict:
     """
@@ -526,16 +597,53 @@ def canonify_df(org_id: str, source_id: str, hours_window: int = 24) -> tuple[pd
 
     # After line: print(f"[canonify] ✅ Pipeline complete in {duration_ms:.2f}ms")
     if not df.empty:
-            # At the end of the canonify pipeline
+        # At the end of the canonify pipeline: ensure Redis keys for entity/industry
+        # are present (defensive) and nudge workers via stream AFTER commit.
+        try:
+            e_key = f"entity:{org_id}:{source_id}"
+            i_key = f"industry:{org_id}:{source_id}"
+            entity_payload = json.dumps({"entity_type": entity_type, "confidence": 1.0})
+            industry_payload = json.dumps({"industry": industry, "confidence": industry_confidence})
+
+            pipe = event_hub.pipeline()
+            if pipe is not None:
+                try:
+                    pipe.setex(e_key, 3600, entity_payload)
+                    pipe.setex(i_key, 3600, industry_payload)
+                    # per-source readiness nudge
+                    msg = json.dumps({"org_id": org_id, "source_id": source_id, "status": "ready"})
+                    pipe.xadd(event_hub.stream_key(org_id, source_id), {"message": msg})
+                    pipe.execute()
+                except Exception as e:
+                    print(f"[canonify] ⚠️ Pipeline nudge failed: {e}")
+                    # Fallback to sequential writes
+                    try:
+                        event_hub.setex(e_key, 3600, entity_payload)
+                        event_hub.setex(i_key, 3600, industry_payload)
+                    except Exception as re:
+                        print(f"[canonify] ❌ Redis setex fallback failed: {re}")
+            else:
+                # Pipeline not available; write sequentially
+                try:
+                    event_hub.setex(e_key, 3600, entity_payload)
+                    event_hub.setex(i_key, 3600, industry_payload)
+                    msg = json.dumps({"org_id": org_id, "source_id": source_id, "status": "ready"})
+                    event_hub.redis.xadd(event_hub.stream_key(org_id, source_id), {"message": msg})
+                except Exception as e:
+                    print(f"[canonify] ❌ Redis nudge failed: {e}")
+
+            # Emit central trigger for worker manager
             try:
-                # Use central event hub to trigger analytics (pubsub + stream)
                 event_hub.emit_analytics_trigger(org_id, source_id, {
                     "type": "kpi_compute",
                     "entity_type": entity_type,
                     "industry": industry,
-                    "rows_inserted": len(df)                  })
+                    "rows_inserted": len(df)
+                })
                 print(f"[canonify] 🚀 Triggered analytics for {source_id}")
             except Exception as e:
                 print(f"[canonify] ⚠️ Analytics trigger failed (non-critical): {e}")
+        except Exception as e:
+            print(f"[canonify] ⚠️ Finalization nudge failed: {e}")
     
     return df, industry, industry_confidence

app/tasks/analytics_worker.py

@@ -313,14 +313,34 @@ class AnalyticsWorker:
         while (time.time() - start) < max_wait:
             entity_key = f"entity:{self.org_id}:{self.source_id}"
             industry_key = f"industry:{self.org_id}:{self.source_id}"
-            
-            if event_hub.exists(entity_key) and event_hub.exists(industry_key):
-                logger.info(f"[WORKER] ✅ Entity & industry keys found")
-                return
-            
+
+            try:
+                # If in-memory cache exists but Redis does not, invalidate cache
+                from app.mapper import _ENTITY_CACHE, _INDUSTRY_CACHE
+
+                cache_ent = _ENTITY_CACHE.get((self.org_id, self.source_id))
+                cache_ind = _INDUSTRY_CACHE.get((self.org_id, self.source_id))
+
+                ent_exists = event_hub.exists(entity_key)
+                ind_exists = event_hub.exists(industry_key)
+
+                if cache_ent and not ent_exists:
+                    _ENTITY_CACHE.pop((self.org_id, self.source_id), None)
+                    logger.debug(f"[WORKER] Cleared stale _ENTITY_CACHE for {self.org_id}/{self.source_id}")
+                if cache_ind and not ind_exists:
+                    _INDUSTRY_CACHE.pop((self.org_id, self.source_id), None)
+                    logger.debug(f"[WORKER] Cleared stale _INDUSTRY_CACHE for {self.org_id}/{self.source_id}")
+
+                if ent_exists and ind_exists:
+                    logger.info(f"[WORKER] ✅ Entity & industry keys found")
+                    return
+
+            except Exception as e:
+                logger.debug(f"[WORKER] Redis/cache check error: {e}")
+
             logger.info(f"[WORKER] ⏳ Waiting for entity/industry keys...")
             await asyncio.sleep(2)
-        
+
         logger.warning(f"[WORKER] ⚠️ Timeout waiting for keys, proceeding anyway")
     # Change _publish() method to use streams
 async def _publish(self, results: Dict[str, Any]):