added strict data verification in cannonify

app/mapper.py

@@ -488,34 +488,47 @@ def canonify_df(org_id: str, source_id: str, hours_window: int = 24) -> tuple[pd
     # ⬇️⬇️⬇️ NEW VERIFICATION STEP ⬇️⬇️⬇️
     # CRITICAL: Ensure data is committed and visible before spawning worker
     
-    print(f"[canonify] 🔍 Verifying data visibility...")
-    
-    # Use a NEW connection to verify data is visible (ensures isolation level is correct)
+    print(f"[canonify] 🔍 Verifying data visibility (expecting {rows_inserted} rows)...")
+
     verification_conn = None
+    MAX_VERIFY_ATTEMPTS = 5
+    VERIFY_DELAY = 1.0  # seconds
+
     try:
-        verification_conn = get_conn(org_id)
-        verification_conn.execute("CHECKPOINT")  # Force any pending writes to flush
+        for attempt in range(MAX_VERIFY_ATTEMPTS):
+            verification_conn = get_conn(org_id)
+            verification_conn.execute("CHECKPOINT")  # Force flush any pending writes
         
-        # Query the table we just inserted into
-        verify_table = f"main.{entity_type}_canonical"
-        verify_count = verification_conn.execute(f"SELECT COUNT(*) FROM {verify_table}").fetchone()[0]
+            # Query the table we just inserted into
+            verify_table = f"main.{entity_type}_canonical"
+            verify_count = verification_conn.execute(f"SELECT COUNT(*) FROM {verify_table}").fetchone()[0]
         
-        if verify_count >= rows_inserted:
-            print(f"[canonify] ✅ Verification passed: {verify_count} rows visible in {verify_table}")
-        else:
-            print(f"[canonify] ❌ Verification failed: Expected {rows_inserted}, found {verify_count} rows")
-            # Don't proceed if verification fails
-            return df, industry, industry_confidence
+            print(f"[canonify] 🔍 Verification attempt {attempt + 1}: {verify_count}/{rows_inserted} rows visible")
+        
+            if verify_count >= rows_inserted:
+                print(f"[canonify] ✅ Verification passed: {verify_count} rows visible in {verify_table}")
+                break
             
+            if attempt < MAX_VERIFY_ATTEMPTS - 1:
+                print(f"[canonify] ⏳ Data not yet visible, waiting {VERIFY_DELAY}s...")
+                try:
+                    verification_conn.close()
+                except Exception:
+                    pass
+                time.sleep(VERIFY_DELAY)
+            else:
+                # This runs if the loop completes without breaking
+                print(f"[canonify] ❌ Verification failed: Expected {rows_inserted} rows, found only {verify_count}")
+                print(f"[canonify] ⚠️ Proceeding anyway but KPI worker may timeout...")
+        
     except Exception as e:
         print(f"[canonify] ❌ Verification error: {e}")
-        return df, industry, industry_confidence
     finally:
         if verification_conn:
             try:
                 verification_conn.close()
                 print("[canonify] 🔒 Verification connection closed")
-            except:
+            except Exception:
                 pass
     # ⬆️⬆️⬆️ END VERIFICATION ⬆️⬆️⬆️
     # 8️⃣ FINAL: Clean DataFrame for response