Merge pull request #137 from R0m1k3/claude/add-product-monitoring-01DyNDGnKBX4fHNSjAkwVckd

fix: Add concurrency limits to prevent Browserless timeout saturation

app/services/search_service.py

@@ -302,7 +302,8 @@ class NewSearchService:
         
         # Phase 2: Scrape details for each result (Parallel)
         # We want to yield results as they complete, not wait for all
-        semaphore = asyncio.Semaphore(3) # Limit concurrency per site
+        # Reduced from 3 to 2 to avoid saturating Browserless
+        semaphore = asyncio.Semaphore(2) # Limit concurrency per site
 
         async def scrape_wrapper(res):
             async with semaphore:
@@ -455,32 +456,37 @@ async def search_products(
     
     # 3. Execute searches and stream results
     # We create a task for each site generator
-    
+
     generators = [NewSearchService.search_site_generator(key, query) for key in site_keys]
-    
+
     # We need to iterate over multiple async generators concurrently
     # This is a bit complex, so we'll use a queue or similar
     # Simpler approach: Use aiostream if available, or just interleave manually
     # For now, let's just run them and yield as we get them.
     # Since we want to show results ASAP, we can use asyncio.as_completed on the *next* item of each generator?
     # No, generators are stateful.
-    
+
     # Simplest robust approach without extra libs:
     # Create a wrapper task for each generator that puts items into a shared Queue
-    
+
     queue = asyncio.Queue()
     active_producers = len(generators)
-    
-    async def producer(gen):
-        try:
-            async for item in gen:
-                await queue.put(item)
-        except Exception as e:
-            logger.error(f"Error in search producer: {e}")
-        finally:
-            await queue.put(None) # Sentinel
 
-    # Start producers
+    # IMPORTANT: Limit concurrent sites to avoid saturating Browserless
+    # Max 2 sites can search in parallel, others wait
+    site_semaphore = asyncio.Semaphore(2)
+
+    async def producer(gen):
+        async with site_semaphore:  # Wait for slot before starting search
+            try:
+                async for item in gen:
+                    await queue.put(item)
+            except Exception as e:
+                logger.error(f"Error in search producer: {e}")
+            finally:
+                await queue.put(None) # Sentinel
+
+    # Start producers (limited by semaphore)
     for gen in generators:
         asyncio.create_task(producer(gen))