Parallel scrape workers (proxy-gated, configurable)

Fan a scrape run out across N workers, each with its own DB session and
its own scraper backend — and since every backend picks a random sticky
session id, each worker scrapes from a distinct residential IP. Dates are
interleaved across workers so each covers a spread of the range. Cuts a
150-date batch from ~30-45 min to ~12-15 min at 3 workers.

- booking_scraper_concurrency config key (default 3); Settings → Scraper
  Proxy has a "Parallel workers" field
- Forced to 1 when the proxy is off (N workers would share one IP and
  hammer it) or when there's a single date
- Both manual and queue paths routed through _scrape_dates_concurrent;
  workers mark their own queue items
- Per-worker rotate-on-block replaces the old global pause-on-block

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

backend/services/booking_scraper.py

@@ -11,16 +11,18 @@ Features:
 - Anti-scrape detection and pause/resume
 """
 
+import asyncio
 import logging
 import threading
 import uuid
 from datetime import date, datetime, timedelta
 from decimal import Decimal
-from typing import List, Optional, Dict, Any
+from typing import List, Optional, Dict, Any, Tuple
 
 from sqlalchemy import text
 from sqlalchemy.orm import Session
 
+from database import SyncSessionLocal
 from .scraper_backends import ScraperBackend, PlaywrightLocalBackend, HotelData, RateData, AvailabilityStatus
 
 logger = logging.getLogger(__name__)
@@ -446,6 +448,114 @@ async def scrape_date(
     }
 
 
+def get_scraper_concurrency(db: Session) -> int:
+    """Number of parallel scrape workers (config key, default 3). Each worker
+    runs its own browser on its own residential proxy IP, so raise this only
+    with proxy IPs and RAM to spare (~0.4 GB per worker)."""
+    row = db.execute(
+        text("SELECT config_value FROM system_config WHERE config_key = 'booking_scraper_concurrency'")
+    ).fetchone()
+    try:
+        return max(1, int(row.config_value)) if row and row.config_value else 3
+    except (ValueError, TypeError):
+        return 3
+
+
+def _effective_concurrency(db: Session, n_jobs: int) -> int:
+    """Clamp configured concurrency to the workload, and force serial when the
+    proxy is off — N workers would share one IP and hammer it, worse than 1."""
+    configured = get_scraper_concurrency(db)
+    if configured <= 1 or n_jobs <= 1:
+        return 1
+    # get_scraper_backend does no I/O beyond the config read; safe to probe.
+    if not get_scraper_backend(db)._proxy_enabled():
+        logger.info("Proxy disabled — running scrape serially (parallelism needs per-worker IPs)")
+        return 1
+    return max(1, min(configured, n_jobs))
+
+
+def _safe_mark_queue(db: Session, queue_id: Optional[int], status: str, error: str = None):
+    """mark_queue_item that never raises — a marking failure shouldn't kill a worker."""
+    if queue_id is None:
+        return
+    try:
+        mark_queue_item(db, queue_id, status, error)
+    except Exception as e:
+        logger.warning(f"Failed to mark queue item {queue_id} as {status}: {e}")
+        try:
+            db.rollback()
+        except Exception:
+            pass
+
+
+async def _scrape_dates_concurrent(
+    jobs: List[Tuple[date, Optional[int]]],
+    config: Dict[str, Any],
+    concurrency: int,
+    batch_id: uuid.UUID,
+) -> Dict[str, int]:
+    """Scrape a list of (date, queue_id) jobs across `concurrency` workers.
+
+    Each worker owns a DB session and a scraper backend — and because every
+    backend picks its own random sticky-session id, each worker scrapes from a
+    distinct residential IP. Dates are interleaved across workers so each covers
+    a spread of the range. Returns aggregate counts."""
+    shards = [jobs[i::concurrency] for i in range(concurrency)]
+    shards = [s for s in shards if s]
+
+    async def worker(shard: List[Tuple[date, Optional[int]]], widx: int) -> Dict[str, int]:
+        acc = {'hotels': 0, 'rates': 0, 'completed': 0, 'failed': 0, 'blocked': 0}
+        wdb = SyncSessionLocal()
+        backend = get_scraper_backend(wdb)
+        try:
+            for rate_date, queue_id in shard:
+                try:
+                    result = await scrape_date(wdb, rate_date, backend, config, batch_id)
+                except Exception as e:
+                    logger.error(f"[worker {widx}] {rate_date} crashed: {e}")
+                    try:
+                        wdb.rollback()
+                    except Exception:
+                        pass
+                    acc['failed'] += 1
+                    _safe_mark_queue(wdb, queue_id, 'failed', str(e))
+                    continue
+
+                if result.get('blocked'):
+                    acc['blocked'] += 1
+                    acc['failed'] += 1
+                    _safe_mark_queue(wdb, queue_id, 'failed', f"Blocked: {result.get('block_reason')}")
+                elif result['success']:
+                    acc['hotels'] += result['hotels_count']
+                    acc['rates'] += result['rates_count']
+                    acc['completed'] += 1
+                    _safe_mark_queue(wdb, queue_id, 'completed')
+                else:
+                    acc['failed'] += 1
+                    _safe_mark_queue(wdb, queue_id, 'failed', result.get('error'))
+        finally:
+            try:
+                await backend.close()
+            except Exception:
+                pass
+            wdb.close()
+        return acc
+
+    results = await asyncio.gather(
+        *(worker(shard, i) for i, shard in enumerate(shards)),
+        return_exceptions=True,
+    )
+
+    agg = {'hotels': 0, 'rates': 0, 'completed': 0, 'failed': 0, 'blocked': 0}
+    for r in results:
+        if isinstance(r, Exception):
+            logger.error(f"Scrape worker crashed: {r}")
+            continue
+        for k in agg:
+            agg[k] += r[k]
+    return agg
+
+
 async def run_manual_scrape(
     db: Session,
     from_date: date,
@@ -501,67 +611,33 @@ async def _run_manual_scrape_locked(
     # Create batch
     batch_id = create_scrape_batch(db, 'manual')
 
-    # Get backend
+    # Build the date list and fan it out across the worker pool
-    backend = get_scraper_backend(db)
+    jobs: List[Tuple[date, Optional[int]]] = []
-
-    total_hotels = 0
-    total_rates = 0
-    dates_completed = 0
-    dates_failed = 0
-
-    try:
     current_date = from_date
     while current_date <= to_date:
-            logger.info(f"Scraping date: {current_date}")
+        jobs.append((current_date, None))
-
-            result = await scrape_date(db, current_date, backend, config, batch_id)
-
-            if result['blocked']:
-                # Blocking detected - pause and exit
-                await set_scraper_paused(db, True, hours=2)
-                update_scrape_batch(
-                    db, batch_id,
-                    status='blocked',
-                    hotels_found=total_hotels,
-                    rates_scraped=total_rates,
-                    error_message=f"Blocked: {result.get('block_reason', 'unknown')}",
-                    blocked=True
-                )
-                return {
-                    'success': False,
-                    'blocked': True,
-                    'block_reason': result.get('block_reason'),
-                    'dates_completed': dates_completed,
-                    'dates_failed': dates_failed,
-                    'hotels_found': total_hotels,
-                    'rates_scraped': total_rates,
-                }
-
-            if result['success']:
-                total_hotels += result['hotels_count']
-                total_rates += result['rates_count']
-                dates_completed += 1
-            else:
-                dates_failed += 1
-                logger.warning(f"Failed to scrape {current_date}: {result.get('error')}")
-
         current_date += timedelta(days=1)
 
-        # Update batch as completed
+    concurrency = _effective_concurrency(db, len(jobs))
+    logger.info(f"Manual scrape {from_date}..{to_date}: {len(jobs)} date(s) across {concurrency} worker(s)")
+
+    try:
+        agg = await _scrape_dates_concurrent(jobs, config, concurrency, batch_id)
+
         update_scrape_batch(
             db, batch_id,
-            status='completed',
+            status='completed' if agg['completed'] or not jobs else 'failed',
-            hotels_found=total_hotels,
+            hotels_found=agg['hotels'],
-            rates_scraped=total_rates
+            rates_scraped=agg['rates'],
         )
 
         return {
-            'success': True,
+            'success': agg['completed'] > 0 or not jobs,
-            'blocked': False,
+            'blocked': agg['blocked'] > 0,
-            'dates_completed': dates_completed,
+            'dates_completed': agg['completed'],
-            'dates_failed': dates_failed,
+            'dates_failed': agg['failed'],
-            'hotels_found': total_hotels,
+            'hotels_found': agg['hotels'],
-            'rates_scraped': total_rates,
+            'rates_scraped': agg['rates'],
         }
 
     except Exception as e:
@@ -569,20 +645,18 @@ async def _run_manual_scrape_locked(
         update_scrape_batch(
             db, batch_id,
             status='failed',
-            hotels_found=total_hotels,
+            hotels_found=0,
-            rates_scraped=total_rates,
+            rates_scraped=0,
             error_message=str(e)
         )
         return {
             'success': False,
             'error': str(e),
-            'dates_completed': dates_completed,
+            'dates_completed': 0,
-            'dates_failed': dates_failed,
+            'dates_failed': len(jobs),
-            'hotels_found': total_hotels,
+            'hotels_found': 0,
-            'rates_scraped': total_rates,
+            'rates_scraped': 0,
         }
-    finally:
-        await backend.close()
 
 
 # ============================================
@@ -737,72 +811,19 @@ async def _process_queue_locked(db: Session) -> Dict[str, Any]:
     )
     db.commit()
 
-    # Get backend
+    # Fan the queue out across the worker pool (each worker marks its own items)
-    backend = get_scraper_backend(db)
+    jobs: List[Tuple[date, Optional[int]]] = [(it['rate_date'], it['id']) for it in items]
-
+    concurrency = _effective_concurrency(db, len(jobs))
-    total_hotels = 0
+    logger.info(f"Queue processing: {len(jobs)} date(s) across {concurrency} worker(s)")
-    total_rates = 0
-    dates_completed = 0
-    dates_failed = 0
 
     try:
-        for item in items:
+        agg = await _scrape_dates_concurrent(jobs, config, concurrency, batch_id)
-            rate_date = item['rate_date']
-            queue_id = item['id']
 
-            logger.info(f"Queue processing: {rate_date} (priority={item['priority']}, attempt={item['attempts']+1})")
-
-            result = await scrape_date(db, rate_date, backend, config, batch_id)
-
-            if result['blocked']:
-                # Mark this item as failed, pause, and stop
-                mark_queue_item(db, queue_id, 'failed', f"Blocked: {result.get('block_reason')}")
-                await set_scraper_paused(db, True, hours=2)
         update_scrape_batch(
             db, batch_id,
-                    status='blocked',
+            status='completed' if agg['completed'] else 'failed',
-                    hotels_found=total_hotels,
+            hotels_found=agg['hotels'],
-                    rates_scraped=total_rates,
+            rates_scraped=agg['rates'],
-                    error_message=f"Blocked: {result.get('block_reason', 'unknown')}",
-                    blocked=True
-                )
-                # Update dates counters
-                db.execute(
-                    text("""
-                        UPDATE booking_scrape_log SET
-                            dates_completed = :completed,
-                            dates_failed = :failed
-                        WHERE batch_id = :bid
-                    """),
-                    {'completed': dates_completed, 'failed': dates_failed + 1, 'bid': str(batch_id)}
-                )
-                db.commit()
-                return {
-                    'success': False,
-                    'blocked': True,
-                    'block_reason': result.get('block_reason'),
-                    'dates_completed': dates_completed,
-                    'dates_failed': dates_failed + 1,
-                    'hotels_found': total_hotels,
-                    'rates_scraped': total_rates,
-                }
-
-            if result['success']:
-                mark_queue_item(db, queue_id, 'completed')
-                total_hotels += result['hotels_count']
-                total_rates += result['rates_count']
-                dates_completed += 1
-            else:
-                mark_queue_item(db, queue_id, 'failed', result.get('error'))
-                dates_failed += 1
-                logger.warning(f"Queue: failed to scrape {rate_date}: {result.get('error')}")
-
-        # Update batch as completed
-        update_scrape_batch(
-            db, batch_id,
-            status='completed',
-            hotels_found=total_hotels,
-            rates_scraped=total_rates
         )
         db.execute(
             text("""
@@ -811,17 +832,17 @@ async def _process_queue_locked(db: Session) -> Dict[str, Any]:
                     dates_failed = :failed
                 WHERE batch_id = :bid
             """),
-            {'completed': dates_completed, 'failed': dates_failed, 'bid': str(batch_id)}
+            {'completed': agg['completed'], 'failed': agg['failed'], 'bid': str(batch_id)}
         )
         db.commit()
 
         return {
-            'success': True,
+            'success': agg['completed'] > 0,
-            'blocked': False,
+            'blocked': agg['blocked'] > 0,
-            'dates_completed': dates_completed,
+            'dates_completed': agg['completed'],
-            'dates_failed': dates_failed,
+            'dates_failed': agg['failed'],
-            'hotels_found': total_hotels,
+            'hotels_found': agg['hotels'],
-            'rates_scraped': total_rates,
+            'rates_scraped': agg['rates'],
         }
 
     except Exception as e:
@@ -829,18 +850,16 @@ async def _process_queue_locked(db: Session) -> Dict[str, Any]:
         update_scrape_batch(
             db, batch_id,
             status='failed',
-            hotels_found=total_hotels,
+            hotels_found=0,
-            rates_scraped=total_rates,
+            rates_scraped=0,
             error_message=str(e)
         )
         return {
             'success': False,
             'error': str(e),
-            'dates_completed': dates_completed,
+            'dates_completed': 0,
-            'dates_failed': dates_failed,
+            'dates_failed': len(jobs),
         }
-    finally:
-        await backend.close()
 
 
 def get_competitor_matrix(

frontend/src/pages/Settings.tsx

@@ -123,6 +123,12 @@ function ProxyTab() {
   const [password, setPassword] = useState('')
   const [testResult, setTestResult] = useState<ProxyTestResult | null>(null)
   const [testError, setTestError] = useState<string | null>(null)
+  const [workers, setWorkers] = useState('3')
+
+  const { data: sysCfg } = useQuery<SystemConfig>({
+    queryKey: ['system-config'],
+    queryFn: () => api.get('/competitors/config/system').then(r => r.data),
+  })
 
   // Seed form once the stored config arrives (password stays blank by design)
   useEffect(() => {
@@ -135,6 +141,17 @@ function ProxyTab() {
     }
   }, [cfg])
 
+  useEffect(() => {
+    if (sysCfg?.booking_scraper_concurrency) setWorkers(String(sysCfg.booking_scraper_concurrency))
+  }, [sysCfg])
+
+  const saveWorkers = useMutation({
+    mutationFn: () => api.post('/competitors/config/system', {
+      key: 'booking_scraper_concurrency', value: String(Math.max(1, parseInt(workers) || 1)),
+    }),
+    onSuccess: () => qc.invalidateQueries({ queryKey: ['system-config'] }),
+  })
+
   const save = useMutation({
     mutationFn: () =>
       api.post('/competitors/config/proxy', {
@@ -248,6 +265,38 @@ function ProxyTab() {
           )}
         </div>
       </div>
+
+      <div className="card">
+        <div className="card-header">
+          <span style={{ display: 'flex', alignItems: 'center', gap: 8 }}>
+            <Network size={16} strokeWidth={1.75} />
+            Parallel Workers
+          </span>
+        </div>
+        <div className="card-body" style={{ display: 'flex', flexDirection: 'column', gap: 12 }}>
+          <p style={{ fontSize: 13, color: 'var(--text-mid)', margin: 0 }}>
+            How many dates to scrape at once. Each worker runs its own browser on its own
+            residential IP, so a run finishes roughly this many times faster. Only takes effect
+            when the proxy is enabled (without it, all workers would share one IP). Budget about
+            0.4&nbsp;GB of RAM per worker — 3 is safe on the current 4&nbsp;GB; raise it after
+            adding RAM.
+          </p>
+          <div style={{ display: 'flex', gap: 8, alignItems: 'center' }}>
+            <input type="number" min={1} max={12} value={workers}
+              onChange={e => setWorkers(e.target.value)} style={{ width: 90 }} />
+            <button className="btn btn-primary btn-sm" onClick={() => saveWorkers.mutate()}
+              disabled={saveWorkers.isPending}>
+              <Save size={13} strokeWidth={1.75} />
+              {saveWorkers.isPending ? 'Saving…' : 'Save'}
+            </button>
+            {!enabled && (
+              <span style={{ fontSize: 12, color: 'var(--warning)' }}>
+                Enable the proxy above for this to apply
+              </span>
+            )}
+          </div>
+        </div>
+      </div>
     </div>
   )
 }
@@ -493,6 +542,7 @@ function SystemTab({ config, isLoading }: { config: SystemConfig | undefined; is
     'booking_proxy_enabled',
     'booking_proxy_host',
     'booking_proxy_country',
+    'booking_scraper_concurrency',
     'sync_newbook_current_rates_enabled',
     'sync_newbook_current_rates_time',
   ]