added dataframe load time so that reddis dont time out

app/tasks/analytics_worker.py

@@ -205,9 +205,12 @@ class AnalyticsWorker:
         return await asyncio.to_thread(self._sync_load_dataframe)
     
     def _sync_load_dataframe(self) -> pd.DataFrame:
-        """Wait for table + data with exponential backoff"""
+        """
+        Load data with resilience: check table, check data, apply filters
+        Returns data even if time window is empty
+        """
         conn = None
-        MAX_WAIT = 30
+        MAX_WAIT = 10  # Reduced from 30s for faster feedback
         INITIAL_RETRY = 0.5
         
         try:
@@ -218,49 +221,65 @@ class AnalyticsWorker:
             
             self._entity_type = entity_info["entity_type"]
             table_name = f"main.{self._entity_type}_canonical"
-            cutoff = datetime.now() - timedelta(hours=self.hours_window)
-            
             conn = get_conn(self.org_id)
             retry_delay = INITIAL_RETRY
             start = time.time()
             
             while (time.time() - start) < MAX_WAIT:
                 try:
-                    count = conn.execute(
-                        f"SELECT COUNT(*) FROM {table_name} WHERE timestamp >= ?",
+                    # Check 1: Does table exist?
+                    table_exists = conn.execute(
+                        "SELECT COUNT(*) FROM information_schema.tables WHERE table_name = ?",
+                        [table_name.split('.')[1]]
+                    ).fetchone()[0] > 0
+                    
+                    if not table_exists:
+                        logger.debug(f"[LOAD] Table {table_name} doesn't exist yet")
+                        time.sleep(retry_delay)
+                        continue
+                    
+                    # Check 2: Does ANY data exist? (no timestamp filter)
+                    count = conn.execute(f"SELECT COUNT(*) FROM {table_name}").fetchone()[0]
+                    
+                    if count == 0:
+                        logger.debug(f"[LOAD] Table exists but is empty, waiting for data...")
+                        time.sleep(retry_delay)
+                        continue
+                    
+                    # Check 3: Try timestamp filter
+                    cutoff = datetime.now() - timedelta(hours=self.hours_window)
+                    df = conn.execute(
+                        f"SELECT * FROM {table_name} WHERE timestamp >= ? ORDER BY timestamp DESC LIMIT 10000",
                         [cutoff]
-                    ).fetchone()[0]
+                    ).df()
+                    
+                    if not df.empty:
+                        logger.info(f"[LOAD] Success: {len(df)} rows × {len(df.columns)} cols (time-filtered)")
+                        return df
                     
-                    if count > 0:
-                        logger.info(f"[LOAD] Table ready: {count} rows (waited {(time.time() - start):.1f}s)")
-                        break
+                    # If timestamp filter is empty, return recent data anyway
+                    logger.warning(f"[LOAD] No data in {self.hours_window}h window, returning recent rows")
+                    df = conn.execute(
+                        f"SELECT * FROM {table_name} ORDER BY timestamp DESC LIMIT 1000"
+                    ).df()
                     
-                    logger.debug(f"[LOAD] Table empty, retrying in {retry_delay}s...")
+                    if not df.empty:
+                        logger.info(f"[LOAD] Success: {len(df)} rows × {len(df.columns)} cols (recent data)")
+                        return df
+                    
+                    time.sleep(retry_delay)
+                    retry_delay = min(retry_delay * 1.5, 5.0)
                     
                 except Exception as e:
-                    if "does not exist" in str(e).lower():
-                        logger.debug(f"[LOAD] Table doesn't exist yet, retrying...")
-                    else:
-                        logger.warning(f"[LOAD] Error: {e}")
-                
-                time.sleep(retry_delay)
-                retry_delay = min(retry_delay * 1.5, 5.0)
-            
-            else:
-                logger.error(f"[LOAD] Timeout after {MAX_WAIT}s")
-                return pd.DataFrame()
-            
-            # Load the data
-            df = conn.execute(
-                f"SELECT * FROM {table_name} WHERE timestamp >= ? ORDER BY timestamp DESC",
-                [cutoff]
-            ).df()
+                    logger.warning(f"[LOAD] Error (retry in {retry_delay}s): {e}")
+                    time.sleep(retry_delay)
+                    retry_delay = min(retry_delay * 1.5, 5.0)
             
-            logger.info(f"[LOAD] Success: {len(df)} rows × {len(df.columns)} cols")
-            return df
+            logger.error(f"[LOAD] Timeout after {MAX_WAIT}s - table exists but no data available")
+            return pd.DataFrame()
             
         except Exception as e:
-            logger.error(f"[LOAD] Fatal: {e}", exc_info=True)
+            logger.error(f"[LOAD] Fatal error: {e}", exc_info=True)
             return pd.DataFrame()
         finally:
             if conn: