mirror of
https://github.com/R0m1k3/Priceflow.git
synced 2026-10-23 23:28:12 +02:00
fix: Add concurrency limits to prevent Browserless timeout saturation
**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.
This commit is contained in:
1 file changed
+21
-15
@@ -302,7 +302,8 @@ class NewSearchService:
|
|||||||
|
|
||||||
# Phase 2: Scrape details for each result (Parallel)
|
# Phase 2: Scrape details for each result (Parallel)
|
||||||
# We want to yield results as they complete, not wait for all
|
# 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 def scrape_wrapper(res):
|
||||||
async with semaphore:
|
async with semaphore:
|
||||||
@@ -455,32 +456,37 @@ async def search_products(
|
|||||||
|
|
||||||
# 3. Execute searches and stream results
|
# 3. Execute searches and stream results
|
||||||
# We create a task for each site generator
|
# We create a task for each site generator
|
||||||
|
|
||||||
generators = [NewSearchService.search_site_generator(key, query) for key in site_keys]
|
generators = [NewSearchService.search_site_generator(key, query) for key in site_keys]
|
||||||
|
|
||||||
# We need to iterate over multiple async generators concurrently
|
# We need to iterate over multiple async generators concurrently
|
||||||
# This is a bit complex, so we'll use a queue or similar
|
# This is a bit complex, so we'll use a queue or similar
|
||||||
# Simpler approach: Use aiostream if available, or just interleave manually
|
# Simpler approach: Use aiostream if available, or just interleave manually
|
||||||
# For now, let's just run them and yield as we get them.
|
# 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?
|
# Since we want to show results ASAP, we can use asyncio.as_completed on the *next* item of each generator?
|
||||||
# No, generators are stateful.
|
# No, generators are stateful.
|
||||||
|
|
||||||
# Simplest robust approach without extra libs:
|
# Simplest robust approach without extra libs:
|
||||||
# Create a wrapper task for each generator that puts items into a shared Queue
|
# Create a wrapper task for each generator that puts items into a shared Queue
|
||||||
|
|
||||||
queue = asyncio.Queue()
|
queue = asyncio.Queue()
|
||||||
active_producers = len(generators)
|
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:
|
for gen in generators:
|
||||||
asyncio.create_task(producer(gen))
|
asyncio.create_task(producer(gen))
|
||||||
|
|
||||||
|
|||||||
Reference in new issue
Block a user