Scraper: process-wide lock, one scrape at a time
Concurrent manual scrapes were interleaving (two Chromium sessions on one LXC) causing the page timeouts behind partial results. SCRAPE_LOCK guards run_manual_scrape and process_queue; the trigger endpoint returns 409 when busy, and the frontend keeps the job queued and retries after 30s. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
parent
bf9425ee7e
commit
b712196262
3 changed files with 56 additions and 3 deletions
|
|
@ -297,6 +297,12 @@ async def trigger_manual_scrape(
|
||||||
if paused_row and paused_row.config_value == 'true':
|
if paused_row and paused_row.config_value == 'true':
|
||||||
raise HTTPException(status_code=400, detail="Scraper is currently paused. Use /unpause first or wait for cooldown.")
|
raise HTTPException(status_code=400, detail="Scraper is currently paused. Use /unpause first or wait for cooldown.")
|
||||||
|
|
||||||
|
# Only one scrape at a time — concurrent Chromium runs cause the page
|
||||||
|
# timeouts that produce partial results
|
||||||
|
from services.booking_scraper import SCRAPE_LOCK
|
||||||
|
if SCRAPE_LOCK.locked():
|
||||||
|
raise HTTPException(status_code=409, detail="A scrape is already running. Try again when it finishes.")
|
||||||
|
|
||||||
# Start background task
|
# Start background task
|
||||||
background_tasks.add_task(run_scrape_sync, from_date, to_date)
|
background_tasks.add_task(run_scrape_sync, from_date, to_date)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -12,6 +12,7 @@ Features:
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
|
import threading
|
||||||
import uuid
|
import uuid
|
||||||
from datetime import date, datetime, timedelta
|
from datetime import date, datetime, timedelta
|
||||||
from decimal import Decimal
|
from decimal import Decimal
|
||||||
|
|
@ -24,6 +25,11 @@ from .scraper_backends import ScraperBackend, PlaywrightLocalBackend, HotelData,
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
# One scrape at a time, process-wide: concurrent Chromium runs starve each
|
||||||
|
# other of memory (page timeouts → partial results) and double the request
|
||||||
|
# rate at Booking.com. Guards both manual scrapes and queue processing.
|
||||||
|
SCRAPE_LOCK = threading.Lock()
|
||||||
|
|
||||||
|
|
||||||
def get_scraper_backend(db: Session) -> ScraperBackend:
|
def get_scraper_backend(db: Session) -> ScraperBackend:
|
||||||
"""
|
"""
|
||||||
|
|
@ -445,6 +451,24 @@ async def run_manual_scrape(
|
||||||
if to_date is None:
|
if to_date is None:
|
||||||
to_date = from_date
|
to_date = from_date
|
||||||
|
|
||||||
|
if not SCRAPE_LOCK.acquire(blocking=False):
|
||||||
|
logger.warning(f"Manual scrape {from_date}..{to_date} refused — another scrape is running")
|
||||||
|
return {
|
||||||
|
'success': False,
|
||||||
|
'already_running': True,
|
||||||
|
'error': 'Another scrape is already running. Try again when it finishes.',
|
||||||
|
}
|
||||||
|
try:
|
||||||
|
return await _run_manual_scrape_locked(db, from_date, to_date)
|
||||||
|
finally:
|
||||||
|
SCRAPE_LOCK.release()
|
||||||
|
|
||||||
|
|
||||||
|
async def _run_manual_scrape_locked(
|
||||||
|
db: Session,
|
||||||
|
from_date: date,
|
||||||
|
to_date: date
|
||||||
|
) -> Dict[str, Any]:
|
||||||
# Check if paused
|
# Check if paused
|
||||||
if await is_scraper_paused(db):
|
if await is_scraper_paused(db):
|
||||||
return {
|
return {
|
||||||
|
|
@ -655,6 +679,20 @@ async def process_queue(db: Session) -> Dict[str, Any]:
|
||||||
Returns:
|
Returns:
|
||||||
Dict with processing results
|
Dict with processing results
|
||||||
"""
|
"""
|
||||||
|
if not SCRAPE_LOCK.acquire(blocking=False):
|
||||||
|
logger.warning("Queue processing skipped — another scrape is running")
|
||||||
|
return {
|
||||||
|
'success': False,
|
||||||
|
'already_running': True,
|
||||||
|
'error': 'Another scrape is already running.',
|
||||||
|
}
|
||||||
|
try:
|
||||||
|
return await _process_queue_locked(db)
|
||||||
|
finally:
|
||||||
|
SCRAPE_LOCK.release()
|
||||||
|
|
||||||
|
|
||||||
|
async def _process_queue_locked(db: Session) -> Dict[str, Any]:
|
||||||
# Check if paused
|
# Check if paused
|
||||||
if await is_scraper_paused(db):
|
if await is_scraper_paused(db):
|
||||||
return {
|
return {
|
||||||
|
|
|
||||||
|
|
@ -949,6 +949,8 @@ const RateMatrixTab: React.FC = () => {
|
||||||
// batch finishes, refetch the matrix, then start the next job.
|
// batch finishes, refetch the matrix, then start the next job.
|
||||||
const [scrapeQueue, setScrapeQueue] = useState<ScrapeJob[]>([])
|
const [scrapeQueue, setScrapeQueue] = useState<ScrapeJob[]>([])
|
||||||
const [scrapeWatch, setScrapeWatch] = useState<{ prevBatch: string | null, startedAt: number, timeoutMs: number } | null>(null)
|
const [scrapeWatch, setScrapeWatch] = useState<{ prevBatch: string | null, startedAt: number, timeoutMs: number } | null>(null)
|
||||||
|
const [dispatchHoldUntil, setDispatchHoldUntil] = useState(0)
|
||||||
|
const [retryTick, setRetryTick] = useState(0)
|
||||||
|
|
||||||
const enqueueScrape = (from: string, to: string) => {
|
const enqueueScrape = (from: string, to: string) => {
|
||||||
setScrapeQueue(q => q.some(j => j.from === from && j.to === to) ? q : [...q, { from, to }])
|
setScrapeQueue(q => q.some(j => j.from === from && j.to === to) ? q : [...q, { from, to }])
|
||||||
|
|
@ -969,18 +971,25 @@ const RateMatrixTab: React.FC = () => {
|
||||||
})
|
})
|
||||||
return (await api.post('/competitors/scrape', { from_date: job.from, to_date: job.to })).data
|
return (await api.post('/competitors/scrape', { from_date: job.from, to_date: job.to })).data
|
||||||
},
|
},
|
||||||
onError: () => {
|
onError: (err: any) => {
|
||||||
setScrapeWatch(null)
|
setScrapeWatch(null)
|
||||||
setScrapeQueue(q => q.slice(1))
|
if (err?.response?.status === 409) {
|
||||||
|
// Server is busy with another scrape — keep the job queued, retry in 30s
|
||||||
|
setDispatchHoldUntil(Date.now() + 30000)
|
||||||
|
setTimeout(() => setRetryTick(t => t + 1), 31000)
|
||||||
|
} else {
|
||||||
|
setScrapeQueue(q => q.slice(1))
|
||||||
|
}
|
||||||
},
|
},
|
||||||
})
|
})
|
||||||
|
|
||||||
// Dispatch the next queued job when idle
|
// Dispatch the next queued job when idle
|
||||||
useEffect(() => {
|
useEffect(() => {
|
||||||
|
if (Date.now() < dispatchHoldUntil) return
|
||||||
if (scrapeQueue.length && !scrapeWatch && !dateScrapeM.isPending) {
|
if (scrapeQueue.length && !scrapeWatch && !dateScrapeM.isPending) {
|
||||||
dateScrapeM.mutate(scrapeQueue[0])
|
dateScrapeM.mutate(scrapeQueue[0])
|
||||||
}
|
}
|
||||||
}, [scrapeQueue, scrapeWatch, dateScrapeM.isPending]) // eslint-disable-line react-hooks/exhaustive-deps
|
}, [scrapeQueue, scrapeWatch, dateScrapeM.isPending, dispatchHoldUntil, retryTick]) // eslint-disable-line react-hooks/exhaustive-deps
|
||||||
|
|
||||||
// Also serves the date-header Booking.com links via location_name
|
// Also serves the date-header Booking.com links via location_name
|
||||||
const { data: watchStatus } = useQuery<ScraperStatus>({
|
const { data: watchStatus } = useQuery<ScraperStatus>({
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue