Scraper: force-reset endpoint + watchdog to prevent stuck lock
Root cause: threading.Lock held indefinitely when Playwright browser hangs inside run_in_executor (finally never fires from the async side). Fixes: - _acquire_scrape_lock/_release_scrape_lock track monotonic timestamp - POST /competitors/scrape/reset force-releases the lock and marks any running batch as interrupted (queue rows stay intact for retry) - GET /competitors/status now includes lock_held_seconds - APScheduler watchdog job every 30 min auto-releases if held >3h - Settings → Scraper Proxy tab shows live lock status (green/amber) with a Force Reset button requiring confirmation Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
d737940a00
commit
e781b1e8b9
4 changed files with 178 additions and 9 deletions
|
|
@ -79,6 +79,7 @@ class ScraperStatusResponse(BaseModel):
|
||||||
location_configured: bool
|
location_configured: bool
|
||||||
location_name: Optional[str]
|
location_name: Optional[str]
|
||||||
last_scrape: Optional[dict]
|
last_scrape: Optional[dict]
|
||||||
|
lock_held_seconds: Optional[int] = None
|
||||||
|
|
||||||
|
|
||||||
# ============================================
|
# ============================================
|
||||||
|
|
@ -133,12 +134,16 @@ async def get_scraper_status(
|
||||||
'error_message': last_scrape_row.error_message,
|
'error_message': last_scrape_row.error_message,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
from services.booking_scraper import get_lock_status
|
||||||
|
lock = get_lock_status()
|
||||||
|
|
||||||
return ScraperStatusResponse(
|
return ScraperStatusResponse(
|
||||||
enabled=config.get('booking_scraper_enabled', 'false') == 'true',
|
enabled=config.get('booking_scraper_enabled', 'false') == 'true',
|
||||||
backend=config.get('booking_scraper_backend', 'playwright_local'),
|
backend=config.get('booking_scraper_backend', 'playwright_local'),
|
||||||
location_configured=location_row is not None,
|
location_configured=location_row is not None,
|
||||||
location_name=location_row.location_name if location_row else None,
|
location_name=location_row.location_name if location_row else None,
|
||||||
last_scrape=last_scrape
|
last_scrape=last_scrape,
|
||||||
|
lock_held_seconds=lock["held_seconds"],
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -408,8 +413,8 @@ async def trigger_manual_scrape(
|
||||||
# Best-effort early 409: the background task re-acquires the lock and will
|
# Best-effort early 409: the background task re-acquires the lock and will
|
||||||
# no-op (logging "another scrape is running") if it loses a millisecond-
|
# no-op (logging "another scrape is running") if it loses a millisecond-
|
||||||
# window race, so no double-run can slip through here.
|
# window race, so no double-run can slip through here.
|
||||||
from services.booking_scraper import SCRAPE_LOCK
|
from services.booking_scraper import get_lock_status
|
||||||
if SCRAPE_LOCK.locked():
|
if get_lock_status()["locked"]:
|
||||||
raise HTTPException(status_code=409, detail="A scrape is already running. Try again when it finishes.")
|
raise HTTPException(status_code=409, detail="A scrape is already running. Try again when it finishes.")
|
||||||
|
|
||||||
# Start background task
|
# Start background task
|
||||||
|
|
@ -423,6 +428,25 @@ async def trigger_manual_scrape(
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/scrape/reset")
|
||||||
|
async def reset_scraper(
|
||||||
|
db: AsyncSession = Depends(get_db),
|
||||||
|
current_user: dict = Depends(get_current_user),
|
||||||
|
):
|
||||||
|
"""Force-release the scrape lock and mark any stuck running batches as interrupted.
|
||||||
|
Use when the scraper is stuck and the 409 won't clear on its own."""
|
||||||
|
if not (current_user.get('is_admin') or 'manage_scraper' in (current_user.get('caps') or [])):
|
||||||
|
raise HTTPException(status_code=403, detail="manage_scraper capability required")
|
||||||
|
|
||||||
|
from services.booking_scraper import force_reset_scraper
|
||||||
|
sync_db = SyncSessionLocal()
|
||||||
|
try:
|
||||||
|
result = force_reset_scraper(sync_db)
|
||||||
|
finally:
|
||||||
|
sync_db.close()
|
||||||
|
return result
|
||||||
|
|
||||||
|
|
||||||
# ============================================
|
# ============================================
|
||||||
# HOTELS MANAGEMENT
|
# HOTELS MANAGEMENT
|
||||||
# ============================================
|
# ============================================
|
||||||
|
|
|
||||||
|
|
@ -68,6 +68,22 @@ async def run_scheduled_parity_check():
|
||||||
await loop.run_in_executor(None, run_parity_check)
|
await loop.run_in_executor(None, run_parity_check)
|
||||||
|
|
||||||
|
|
||||||
|
async def run_scrape_watchdog():
|
||||||
|
"""Auto-release the scrape lock if held for >3 hours (hung Playwright browser)."""
|
||||||
|
from services.booking_scraper import get_lock_status, force_reset_scraper
|
||||||
|
status = get_lock_status()
|
||||||
|
held = status.get("held_seconds")
|
||||||
|
if held and held > 3 * 3600:
|
||||||
|
logger.warning(f"Scrape watchdog: lock held for {held}s — force releasing")
|
||||||
|
db = SyncSessionLocal()
|
||||||
|
try:
|
||||||
|
force_reset_scraper(db)
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f"Scrape watchdog reset failed: {e}")
|
||||||
|
finally:
|
||||||
|
db.close()
|
||||||
|
|
||||||
|
|
||||||
async def run_scheduled_booking_scrape_async():
|
async def run_scheduled_booking_scrape_async():
|
||||||
from jobs.scrape_booking_rates import run_scheduled_booking_scrape
|
from jobs.scrape_booking_rates import run_scheduled_booking_scrape
|
||||||
import asyncio
|
import asyncio
|
||||||
|
|
@ -131,8 +147,17 @@ def start_scheduler():
|
||||||
replace_existing=True,
|
replace_existing=True,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# Scrape lock watchdog — every 30 min; force-releases if held >3 hours
|
||||||
|
from apscheduler.triggers.interval import IntervalTrigger
|
||||||
|
scheduler.add_job(
|
||||||
|
run_scrape_watchdog,
|
||||||
|
IntervalTrigger(minutes=30),
|
||||||
|
id='scrape_watchdog',
|
||||||
|
replace_existing=True,
|
||||||
|
)
|
||||||
|
|
||||||
scheduler.start()
|
scheduler.start()
|
||||||
logger.info(f"Scheduler started: booking scrape at {scrape_hour:02d}:{scrape_minute:02d}, rates fetch at {rates_hour:02d}:{rates_minute:02d}, direct scrape at 06:00, parity check at 06:45")
|
logger.info(f"Scheduler started: booking scrape at {scrape_hour:02d}:{scrape_minute:02d}, rates fetch at {rates_hour:02d}:{rates_minute:02d}, direct scrape at 06:00, parity check at 06:45, watchdog every 30m")
|
||||||
|
|
||||||
|
|
||||||
def shutdown_scheduler():
|
def shutdown_scheduler():
|
||||||
|
|
|
||||||
|
|
@ -14,6 +14,7 @@ Features:
|
||||||
import asyncio
|
import asyncio
|
||||||
import logging
|
import logging
|
||||||
import threading
|
import threading
|
||||||
|
import time
|
||||||
import uuid
|
import uuid
|
||||||
from datetime import date, timedelta
|
from datetime import date, timedelta
|
||||||
from decimal import Decimal
|
from decimal import Decimal
|
||||||
|
|
@ -32,6 +33,48 @@ logger = logging.getLogger(__name__)
|
||||||
# other of memory (page timeouts → partial results) and double the request
|
# other of memory (page timeouts → partial results) and double the request
|
||||||
# rate at Booking.com. Guards both manual scrapes and queue processing.
|
# rate at Booking.com. Guards both manual scrapes and queue processing.
|
||||||
SCRAPE_LOCK = threading.Lock()
|
SCRAPE_LOCK = threading.Lock()
|
||||||
|
_lock_acquired_at: Optional[float] = None # monotonic timestamp when lock was acquired
|
||||||
|
|
||||||
|
|
||||||
|
def _acquire_scrape_lock() -> bool:
|
||||||
|
global _lock_acquired_at
|
||||||
|
if SCRAPE_LOCK.acquire(blocking=False):
|
||||||
|
_lock_acquired_at = time.monotonic()
|
||||||
|
return True
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
def _release_scrape_lock():
|
||||||
|
global _lock_acquired_at
|
||||||
|
_lock_acquired_at = None
|
||||||
|
try:
|
||||||
|
SCRAPE_LOCK.release()
|
||||||
|
except RuntimeError:
|
||||||
|
pass # already released — race between watchdog and normal release
|
||||||
|
|
||||||
|
|
||||||
|
def get_lock_status() -> dict:
|
||||||
|
"""Return lock state and how long it has been held (seconds)."""
|
||||||
|
locked = SCRAPE_LOCK.locked()
|
||||||
|
held = int(time.monotonic() - _lock_acquired_at) if (locked and _lock_acquired_at) else None
|
||||||
|
return {"locked": locked, "held_seconds": held}
|
||||||
|
|
||||||
|
|
||||||
|
def force_reset_scraper(db: Session) -> dict:
|
||||||
|
"""Force-release the scrape lock and tidy up any stuck DB state.
|
||||||
|
Safe to call even when the lock is not held."""
|
||||||
|
was_locked = SCRAPE_LOCK.locked()
|
||||||
|
_release_scrape_lock()
|
||||||
|
# Mark any batch left in 'running' state as interrupted
|
||||||
|
db.execute(text("""
|
||||||
|
UPDATE booking_scrape_log
|
||||||
|
SET status = 'interrupted', completed_at = NOW(),
|
||||||
|
error_message = 'Force-reset by admin'
|
||||||
|
WHERE status = 'running'
|
||||||
|
"""))
|
||||||
|
db.commit()
|
||||||
|
logger.warning(f"Scrape lock force-reset (was_locked={was_locked})")
|
||||||
|
return {"was_locked": was_locked, "reset": True}
|
||||||
|
|
||||||
|
|
||||||
def get_scraper_backend(db: Session) -> ScraperBackend:
|
def get_scraper_backend(db: Session) -> ScraperBackend:
|
||||||
|
|
@ -509,7 +552,7 @@ 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):
|
if not _acquire_scrape_lock():
|
||||||
logger.warning(f"Manual scrape {from_date}..{to_date} refused — another scrape is running")
|
logger.warning(f"Manual scrape {from_date}..{to_date} refused — another scrape is running")
|
||||||
return {
|
return {
|
||||||
'success': False,
|
'success': False,
|
||||||
|
|
@ -519,7 +562,7 @@ async def run_manual_scrape(
|
||||||
try:
|
try:
|
||||||
return await _run_manual_scrape_locked(db, from_date, to_date)
|
return await _run_manual_scrape_locked(db, from_date, to_date)
|
||||||
finally:
|
finally:
|
||||||
SCRAPE_LOCK.release()
|
_release_scrape_lock()
|
||||||
|
|
||||||
|
|
||||||
async def _run_manual_scrape_locked(
|
async def _run_manual_scrape_locked(
|
||||||
|
|
@ -694,7 +737,7 @@ 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):
|
if not _acquire_scrape_lock():
|
||||||
logger.warning("Queue processing skipped — another scrape is running")
|
logger.warning("Queue processing skipped — another scrape is running")
|
||||||
return {
|
return {
|
||||||
'success': False,
|
'success': False,
|
||||||
|
|
@ -704,7 +747,7 @@ async def process_queue(db: Session) -> Dict[str, Any]:
|
||||||
try:
|
try:
|
||||||
return await _process_queue_locked(db)
|
return await _process_queue_locked(db)
|
||||||
finally:
|
finally:
|
||||||
SCRAPE_LOCK.release()
|
_release_scrape_lock()
|
||||||
|
|
||||||
|
|
||||||
async def _process_queue_locked(db: Session) -> Dict[str, Any]:
|
async def _process_queue_locked(db: Session) -> Dict[str, Any]:
|
||||||
|
|
|
||||||
|
|
@ -1,7 +1,7 @@
|
||||||
import { useState, useEffect } from 'react'
|
import { useState, useEffect } from 'react'
|
||||||
import { useParams, useNavigate } from 'react-router-dom'
|
import { useParams, useNavigate } from 'react-router-dom'
|
||||||
import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query'
|
import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query'
|
||||||
import { Save, RefreshCw, Database, Clock, BedDouble, ChevronUp, ChevronDown, Network, ShieldCheck } from 'lucide-react'
|
import { Save, RefreshCw, Database, Clock, BedDouble, ChevronUp, ChevronDown, Network, ShieldCheck, AlertTriangle } from 'lucide-react'
|
||||||
import api from '../api'
|
import api from '../api'
|
||||||
|
|
||||||
const TABS = [
|
const TABS = [
|
||||||
|
|
@ -488,6 +488,83 @@ function ProxyTab() {
|
||||||
</div>
|
</div>
|
||||||
</div>
|
</div>
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
|
<ScraperLockCard />
|
||||||
|
</div>
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
// ─── Scraper Lock Status Card ─────────────────────────────────────────────────
|
||||||
|
|
||||||
|
function ScraperLockCard() {
|
||||||
|
const qc = useQueryClient()
|
||||||
|
|
||||||
|
const { data: status, isLoading } = useQuery<{ lock_held_seconds: number | null }>({
|
||||||
|
queryKey: ['scraper-status-lock'],
|
||||||
|
queryFn: () => api.get('/competitors/status').then(r => r.data),
|
||||||
|
refetchInterval: 15000,
|
||||||
|
})
|
||||||
|
|
||||||
|
const resetM = useMutation({
|
||||||
|
mutationFn: () => api.post('/competitors/scrape/reset').then(r => r.data),
|
||||||
|
onSuccess: () => qc.invalidateQueries({ queryKey: ['scraper-status-lock'] }),
|
||||||
|
})
|
||||||
|
|
||||||
|
const held = status?.lock_held_seconds ?? null
|
||||||
|
const isStuck = held !== null && held > 0
|
||||||
|
const heldStr = held
|
||||||
|
? held >= 3600
|
||||||
|
? `${Math.floor(held / 3600)}h ${Math.floor((held % 3600) / 60)}m`
|
||||||
|
: held >= 60
|
||||||
|
? `${Math.floor(held / 60)}m ${held % 60}s`
|
||||||
|
: `${held}s`
|
||||||
|
: null
|
||||||
|
|
||||||
|
return (
|
||||||
|
<div className="card" style={{ marginTop: 20 }}>
|
||||||
|
<div className="card-header">Scraper Lock</div>
|
||||||
|
<div style={{ padding: '16px 20px' }}>
|
||||||
|
<div style={{ display: 'flex', alignItems: 'center', gap: 12, marginBottom: 12 }}>
|
||||||
|
<span style={{
|
||||||
|
display: 'inline-flex', alignItems: 'center', gap: 6,
|
||||||
|
padding: '4px 10px', borderRadius: 20, fontSize: 13, fontWeight: 600,
|
||||||
|
background: isStuck ? '#fef3c7' : '#dcfce7',
|
||||||
|
color: isStuck ? '#d97706' : '#16a34a',
|
||||||
|
}}>
|
||||||
|
{isStuck
|
||||||
|
? <><AlertTriangle size={13} strokeWidth={1.75} /> Locked ({heldStr})</>
|
||||||
|
: <><ShieldCheck size={13} strokeWidth={1.75} /> Idle</>}
|
||||||
|
</span>
|
||||||
|
{isStuck && (
|
||||||
|
<span style={{ fontSize: 12, color: 'var(--text-mid)' }}>
|
||||||
|
A scrape may be stuck. The watchdog will auto-release after 3 hours.
|
||||||
|
</span>
|
||||||
|
)}
|
||||||
|
</div>
|
||||||
|
<p style={{ fontSize: 13, color: 'var(--text-mid)', marginBottom: 12 }}>
|
||||||
|
Force-release the scrape lock if it is stuck (e.g. a hung Playwright browser).
|
||||||
|
Any in-progress scrape batch will be marked interrupted; the queue remains intact
|
||||||
|
and will retry on the next scheduled run or manual trigger.
|
||||||
|
</p>
|
||||||
|
<button
|
||||||
|
className="btn btn-sm"
|
||||||
|
style={{ background: '#fee2e2', color: '#dc2626', border: '1px solid #fca5a5' }}
|
||||||
|
disabled={resetM.isPending || isLoading}
|
||||||
|
onClick={() => {
|
||||||
|
if (window.confirm('Force-reset the scraper lock? Any running scrape will be interrupted.')) {
|
||||||
|
resetM.mutate()
|
||||||
|
}
|
||||||
|
}}
|
||||||
|
>
|
||||||
|
<RefreshCw size={13} strokeWidth={1.75} style={resetM.isPending ? { animation: 'spin 1.5s linear infinite' } : undefined} />
|
||||||
|
{resetM.isPending ? 'Resetting…' : 'Force Reset Lock'}
|
||||||
|
</button>
|
||||||
|
{resetM.isSuccess && (
|
||||||
|
<span style={{ marginLeft: 12, fontSize: 12, color: 'var(--success)' }}>
|
||||||
|
✓ Reset — was {(resetM.data as any)?.was_locked ? 'locked' : 'already idle'}
|
||||||
|
</span>
|
||||||
|
)}
|
||||||
|
</div>
|
||||||
</div>
|
</div>
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue