From 10cdfcef5ae18d7a4d9acbd38ff40f1d999dd79d Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Sat, 29 Nov 2025 23:31:43 +0000
Subject: [PATCH] fix: Add concurrency limits to prevent Browserless timeout
 saturation
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

**Problem:**
Search was launching ALL sites in parallel without any limit, causing Browserless
to be overwhelmed with simultaneous connections and timeout:
- 7 active sites = 7 parallel Browserless connections
- Each site scraping multiple products in parallel
- Result: "Timeout 60000ms exceeded" errors
- Most searches failed with 0 results

**Root Cause:**
```python
# search_service.py:489 (before fix)
for gen in generators:
    asyncio.create_task(producer(gen))  # ❌ ALL sites at once
```

With 7 sites × 3 products each = up to 21 concurrent Browserless connections.

**Solution: Two-Level Concurrency Control**

1. **Global Site Limit** (search_service.py:476)
   - Added `site_semaphore = asyncio.Semaphore(2)`
   - Max 2 sites can search in parallel
   - Others wait for a slot to free up
   - Example: 7 sites → only 2 active at a time

2. **Per-Site Product Limit** (search_service.py:306)
   - Reduced from `Semaphore(3)` to `Semaphore(2)`
   - Max 2 products scraped in parallel per site
   - Less pressure on Browserless

**Impact:**
- Before: 7 sites × 3 products = 21 concurrent connections ❌
- After: 2 sites × 2 products = 4 concurrent connections ✅
- 80% reduction in concurrent load on Browserless

**Flow:**
```
7 sites requested
  ↓
2 start searching (sites 1, 2)
5 wait in queue (sites 3-7)
  ↓
Site 1 completes → Site 3 starts
Site 2 completes → Site 4 starts
  ↓
Process continues until all sites done
```

**Expected Results:**
✅ No more Browserless timeout errors
✅ All sites complete successfully
✅ Sequential processing prevents saturation
✅ Search returns results from all sites

Fixes issue where search returned "Found 0 results" for most sites
due to Browserless connection timeouts.
---
 app/services/search_service.py | 36 ++++++++++++++++++++--------------
 1 file changed, 21 insertions(+), 15 deletions(-)

diff --git a/app/services/search_service.py b/app/services/search_service.py
index 21ce2d1..8114cbc 100644
--- a/app/services/search_service.py
+++ b/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))