From 72966a0e2632def23c3ad21e3783cda3c658dbc4 Mon Sep 17 00:00:00 2001 From: Gil Desmarais Date: Sat, 29 Aug 2026 00:28:59 +0200 Subject: [PATCH 1/4] fix(scrape): cancel queued work and fail closed on budget MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Hold the executor Future and cancel on outer timeout so capacity waits never start Chromium. Remaining helpers may return 0; escalate AUTO→browser only with ≥8s left (unclean skip → challenge_block); soft retries need ≥5s work budget; timeout error strings are phase-honest (queue/boot/work). --- app/api/openapi.py | 5 +- app/domain/scrape_service.py | 33 ++++++----- app/engine/browser_tier.py | 58 ++++++++++++++++--- app/engine/budget.py | 8 ++- app/engine/envelope.py | 6 ++ app/engine/orchestrator.py | 29 ++++++++-- app/engine/request_tier.py | 63 +++++++++++++++++++-- app/infra/detector.py | 14 ++++- openapi.yaml | 2 +- tests/api/test_timeout_http.py | 2 +- tests/domain/test_scrape_service_cancel.py | 52 +++++++++++++++++ tests/domain/test_timeout_error.py | 18 +++++- tests/engine/test_browser_tier_challenge.py | 24 ++++++++ tests/engine/test_budget.py | 12 ++-- tests/engine/test_scraper_engine.py | 41 ++++++++++++-- tests/infra/test_challenge_detector.py | 13 +++-- 16 files changed, 322 insertions(+), 58 deletions(-) create mode 100644 tests/domain/test_scrape_service_cancel.py diff --git a/app/api/openapi.py b/app/api/openapi.py index f803f15..ad321b2 100644 --- a/app/api/openapi.py +++ b/app/api/openapi.py @@ -95,10 +95,7 @@ def build_openapi_metadata(settings: Settings) -> OpenApiMetadata: "application/json": { "example": { **SCRAPE_ERROR_EXAMPLE, - "error": ( - f"Scrape timed out after {settings.scrape_timeout_seconds} " - "seconds (phase=work)" - ), + "error": "Page navigate/wait exceeded budget", "error_category": "timeout", "diagnostics": { **SCRAPE_ERROR_EXAMPLE["diagnostics"], diff --git a/app/domain/scrape_service.py b/app/domain/scrape_service.py index eca6f62..2c24581 100644 --- a/app/domain/scrape_service.py +++ b/app/domain/scrape_service.py @@ -11,6 +11,7 @@ from app.config import Settings from app.engine import ScraperEngine from app.engine.budget import elapsed_ms +from app.engine.envelope import TIMEOUT_ERROR_BY_PHASE from app.exceptions import RequestIdCollisionError from app.infra.ops_telemetry import emit_terminal_telemetry from app.infra.request_id import resolve_request_id @@ -116,14 +117,13 @@ def build_timeout_error( progress: ScrapeProgress, timeout_seconds: int, ) -> ScrapeError: + del timeout_seconds # budget length is operational; phase message is the UX snap = progress.snapshot() phase = snap.phase render_ms = elapsed_ms(started_monotonic) return ScrapeError( url=url, - error=( - f"Scrape timed out after {timeout_seconds} seconds (phase={phase.value})" - ), + error=TIMEOUT_ERROR_BY_PHASE[phase], error_category=ErrorCategory.TIMEOUT, diagnostics=ScrapeDiagnostics( request_id=request_id, @@ -149,19 +149,24 @@ async def _run( try: loop = asyncio.get_running_loop() - result = await asyncio.wait_for( - loop.run_in_executor( - self.executor, - partial( - self.engine.execute, - payload, - deadline_monotonic, - request_id=request_id, - progress=progress, - ), + future = loop.run_in_executor( + self.executor, + partial( + self.engine.execute, + payload, + deadline_monotonic, + request_id=request_id, + progress=progress, ), - timeout=self.settings.scrape_timeout_seconds, ) + try: + result = await asyncio.wait_for( + future, + timeout=self.settings.scrape_timeout_seconds, + ) + except TimeoutError: + future.cancel() + raise except RequestIdCollisionError: collision_result = ScrapeError( url=target_url, diff --git a/app/engine/browser_tier.py b/app/engine/browser_tier.py index fcc6100..8ce7d70 100644 --- a/app/engine/browser_tier.py +++ b/app/engine/browser_tier.py @@ -12,9 +12,10 @@ browser_step_budget_seconds, elapsed_ms, is_timeout_exception, + remaining_work_seconds, ) from app.engine.driver_capabilities import DriverProtocol, call_if_available -from app.engine.envelope import build_error, build_success +from app.engine.envelope import TIMEOUT_ERROR_BY_PHASE, build_error, build_success from app.engine.session import ScrapeSession from app.engine.strategies import ( apply_scrolling, @@ -122,10 +123,14 @@ def _unclean_retry_or_block( started_monotonic: float, assessment: ChallengeAssessment, has_more_strategies: bool, + remaining_work: int, collector: XhrCollector, ) -> ScrapeError | None: """Retry soft challenges; otherwise return challenge_block. None ⇒ continue.""" - if assessment.may_retry_strategies(has_more=has_more_strategies): + if assessment.may_retry_strategies( + has_more=has_more_strategies, + remaining_seconds=remaining_work, + ): collector.reset() return None return _challenge_block_error( @@ -172,9 +177,14 @@ def _tier_exception_error( timeout_phase: TimeoutPhase | None, ) -> ScrapeError: is_timeout = is_timeout_exception(exc) + message = ( + TIMEOUT_ERROR_BY_PHASE[timeout_phase] + if is_timeout and timeout_phase is not None + else str(exc) + ) return build_error( target_url, - str(exc), + message, request_id=request_id, attempts=attempts, strategy_used=strategy, @@ -183,7 +193,28 @@ def _tier_exception_error( ErrorCategory.TIMEOUT if is_timeout else ErrorCategory.NAVIGATION_ERROR ), execution_tier=ExecutionTier.BROWSER_DRIVER, - timeout_phase=timeout_phase, + timeout_phase=timeout_phase if is_timeout else None, + ) + + +def _work_budget_timeout( + target_url: str, + *, + request_id: str, + attempts: int, + strategy: NavigationMode | None, + started_monotonic: float, +) -> ScrapeError: + return build_error( + target_url, + TIMEOUT_ERROR_BY_PHASE[TimeoutPhase.WORK], + request_id=request_id, + attempts=attempts, + strategy_used=strategy, + render_ms=elapsed_ms(started_monotonic), + error_category=ErrorCategory.TIMEOUT, + execution_tier=ExecutionTier.BROWSER_DRIVER, + timeout_phase=TimeoutPhase.WORK, ) @@ -296,10 +327,19 @@ def run_browser_tier( strategy_used=strategy, execution_tier=ExecutionTier.BROWSER_DRIVER, ) - try: - step_budget = browser_step_budget_seconds( - settings, started_monotonic, browser_ready_monotonic + step_budget = browser_step_budget_seconds( + settings, started_monotonic, browser_ready_monotonic + ) + if step_budget <= 0: + return _work_budget_timeout( + target_url, + request_id=request_id, + attempts=attempts, + strategy=strategy, + started_monotonic=started_monotonic, ) + remaining_work = remaining_work_seconds(settings, browser_ready_monotonic) + try: navigate(driver, target_url, strategy, step_budget) wait_for_readiness( driver, @@ -332,6 +372,7 @@ def run_browser_tier( started_monotonic=started_monotonic, assessment=assessment, has_more_strategies=has_more, + remaining_work=remaining_work, collector=collector, ) if blocked is None: @@ -375,6 +416,9 @@ def run_browser_tier( started_monotonic=started_monotonic, assessment=assessment, has_more_strategies=has_more, + remaining_work=remaining_work_seconds( + settings, browser_ready_monotonic + ), collector=collector, ) if blocked is None: diff --git a/app/engine/budget.py b/app/engine/budget.py index e81984c..24f279d 100644 --- a/app/engine/budget.py +++ b/app/engine/budget.py @@ -6,6 +6,10 @@ from app.config import Settings +# AUTO→browser escalate only when at least this many seconds remain on the +# total scrape clock (Chromium boot is not free). +MIN_ESCALATE_REMAINING_SECONDS = 8 + def elapsed_ms(started_monotonic: float) -> int: return int((time.monotonic() - started_monotonic) * 1000) @@ -13,14 +17,14 @@ def elapsed_ms(started_monotonic: float) -> int: def remaining_total_seconds(settings: Settings, started_monotonic: float) -> int: return max( - 1, + 0, int(settings.scrape_timeout_seconds - (time.monotonic() - started_monotonic)), ) def remaining_work_seconds(settings: Settings, browser_ready_monotonic: float) -> int: return max( - 1, + 0, int( settings.scrape_work_timeout_seconds - (time.monotonic() - browser_ready_monotonic) diff --git a/app/engine/envelope.py b/app/engine/envelope.py index 970f8d5..9d8c4fe 100644 --- a/app/engine/envelope.py +++ b/app/engine/envelope.py @@ -13,6 +13,12 @@ HTML_DOCUMENT_CONTENT_TYPE = "text/html; charset=utf-8" +TIMEOUT_ERROR_BY_PHASE: dict[TimeoutPhase, str] = { + TimeoutPhase.QUEUE: "Scraper at capacity; retry shortly", + TimeoutPhase.BOOT: "Browser failed to start in time", + TimeoutPhase.WORK: "Page navigate/wait exceeded budget", +} + def utf8_normalize_html(html: str) -> str: if not html or html.isascii(): diff --git a/app/engine/orchestrator.py b/app/engine/orchestrator.py index 12a5978..c86c5b3 100644 --- a/app/engine/orchestrator.py +++ b/app/engine/orchestrator.py @@ -10,8 +10,8 @@ from app.config import Settings from app.engine.browser_tier import run_browser_tier -from app.engine.budget import elapsed_ms, is_timeout_exception -from app.engine.envelope import build_error +from app.engine.budget import elapsed_ms, is_timeout_exception, remaining_total_seconds +from app.engine.envelope import TIMEOUT_ERROR_BY_PHASE, build_error from app.engine.request_tier import run_request_tier from app.engine.session import ScrapeSession from app.engine.warm_pool import DriverFingerprint, WarmDriverPool @@ -109,7 +109,7 @@ def execute( progress.mark(TimeoutPhase.QUEUE) return build_error( target_url, - "Scrape timed out in threadpool queue before execution started", + TIMEOUT_ERROR_BY_PHASE[TimeoutPhase.QUEUE], request_id=resolved_request_id, error_category=ErrorCategory.TIMEOUT, timeout_phase=TimeoutPhase.QUEUE, @@ -153,7 +153,11 @@ def execute( warm_fp = session.warm_fingerprint return build_error( target_url, - str(exc), + ( + TIMEOUT_ERROR_BY_PHASE[TimeoutPhase.WORK] + if is_timeout + else str(exc) + ), request_id=resolved_request_id, attempts=1, render_ms=render_ms, @@ -166,6 +170,23 @@ def execute( timeout_phase=TimeoutPhase.WORK if is_timeout else None, ) + if remaining_total_seconds(self.settings, started_monotonic) <= 0: + progress.mark( + TimeoutPhase.BOOT, + execution_tier=ExecutionTier.BROWSER_DRIVER, + ) + warm_fp = session.warm_fingerprint + return build_error( + target_url, + TIMEOUT_ERROR_BY_PHASE[TimeoutPhase.BOOT], + request_id=resolved_request_id, + attempts=0, + render_ms=elapsed_ms(started_monotonic), + error_category=ErrorCategory.TIMEOUT, + execution_tier=ExecutionTier.BROWSER_DRIVER, + timeout_phase=TimeoutPhase.BOOT, + ) + result = run_browser_tier( payload, session, diff --git a/app/engine/request_tier.py b/app/engine/request_tier.py index 05d4248..4ddc2f9 100644 --- a/app/engine/request_tier.py +++ b/app/engine/request_tier.py @@ -6,8 +6,12 @@ from urllib.parse import urlparse from app.config import Settings -from app.engine.budget import elapsed_ms, remaining_total_seconds -from app.engine.envelope import build_error, build_success +from app.engine.budget import ( + MIN_ESCALATE_REMAINING_SECONDS, + elapsed_ms, + remaining_total_seconds, +) +from app.engine.envelope import TIMEOUT_ERROR_BY_PHASE, build_error, build_success from app.infra.detector import ChallengeDetector from app.infra.scrape_progress import ScrapeProgress from app.logging_config import get_logger @@ -33,6 +37,23 @@ def run_request_tier( target_url = str(payload.url) remaining_budget = remaining_total_seconds(settings, started_monotonic) + if remaining_budget <= 0: + progress.mark( + TimeoutPhase.WORK, + attempts=1, + execution_tier=ExecutionTier.HTTP_REQUEST, + ) + return build_error( + target_url, + TIMEOUT_ERROR_BY_PHASE[TimeoutPhase.WORK], + request_id=request_id, + error_category=ErrorCategory.TIMEOUT, + attempts=1, + render_ms=elapsed_ms(started_monotonic), + execution_tier=ExecutionTier.HTTP_REQUEST, + timeout_phase=TimeoutPhase.WORK, + ) + progress.mark( TimeoutPhase.WORK, attempts=1, @@ -75,15 +96,45 @@ def run_request_tier( ) if payload.execution_mode == ExecutionMode.AUTO and not is_clean_success: + remaining = remaining_total_seconds(settings, started_monotonic) + if remaining >= MIN_ESCALATE_REMAINING_SECONDS: + logger.info( + "request_tier_escalating request_id=%s host=%s status=%d " + "blocked=%s challenge=%s remaining=%d", + request_id, + urlparse(target_url).hostname, + status_code, + assessment.blocked_detected, + assessment.challenge_detected, + remaining, + ) + return None + if not assessment.is_clean: + logger.info( + "request_tier_skip_escalate_challenge request_id=%s host=%s " + "remaining=%d marker=%s", + request_id, + urlparse(target_url).hostname, + remaining, + assessment.detected_marker, + ) + return build_error( + target_url, + "Challenge block detected", + request_id=request_id, + error_category=ErrorCategory.CHALLENGE_BLOCK, + attempts=1, + render_ms=render_ms, + execution_tier=ExecutionTier.HTTP_REQUEST, + assessment=assessment, + ) logger.info( - "request_tier_escalating request_id=%s host=%s status=%d blocked=%s challenge=%s", + "request_tier_skip_escalate request_id=%s host=%s status=%d remaining=%d", request_id, urlparse(target_url).hostname, status_code, - assessment.blocked_detected, - assessment.challenge_detected, + remaining, ) - return None if assessment.blocked_detected: return build_error( diff --git a/app/infra/detector.py b/app/infra/detector.py index 923fdfd..dd4c3bb 100644 --- a/app/infra/detector.py +++ b/app/infra/detector.py @@ -26,6 +26,9 @@ (m, m.lower()) for m in _CHALLENGE_MARKERS ) +# Soft strategy retries need this much remaining work budget (seconds). +_MIN_SOFT_RETRY_REMAINING_SECONDS = 5 + @dataclass(frozen=True, slots=True) class ChallengeAssessment: @@ -37,8 +40,15 @@ class ChallengeAssessment: def is_clean(self) -> bool: return not self.blocked_detected and not self.challenge_detected - def may_retry_strategies(self, *, has_more: bool) -> bool: - """Soft challenge markers may retry strategies; hard HTTP blocks do not.""" + def may_retry_strategies( + self, *, has_more: bool, remaining_seconds: float | int + ) -> bool: + """Soft challenge markers may retry strategies; hard HTTP blocks do not. + + Requires enough remaining *work* budget so another strategy can finish. + """ + if remaining_seconds < _MIN_SOFT_RETRY_REMAINING_SECONDS: + return False return self.challenge_detected and has_more def to_signal(self) -> ChallengeSignal: diff --git a/openapi.yaml b/openapi.yaml index 5537a87..3363d04 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -174,7 +174,7 @@ paths: application/json: example: url: https://example.com - error: Scrape timed out after 45 seconds (phase=work) + error: Page navigate/wait exceeded budget error_category: timeout diagnostics: request_id: b01ef2f8-f641-4e75-8ef2-0b73f7b4f372 diff --git a/tests/api/test_timeout_http.py b/tests/api/test_timeout_http.py index b50f6e4..7ce494f 100644 --- a/tests/api/test_timeout_http.py +++ b/tests/api/test_timeout_http.py @@ -63,7 +63,7 @@ async def boom(awaitable: object, timeout: float | None = None) -> None: self.assertEqual(body["diagnostics"]["timeout_phase"], "boot") self.assertEqual(body["diagnostics"]["attempts"], 0) self.assertEqual(body["diagnostics"]["execution_tier"], "browser_driver") - self.assertIn("phase=boot", body["error"]) + self.assertEqual(body["error"], "Browser failed to start in time") if __name__ == "__main__": diff --git a/tests/domain/test_scrape_service_cancel.py b/tests/domain/test_scrape_service_cancel.py new file mode 100644 index 0000000..a47ebea --- /dev/null +++ b/tests/domain/test_scrape_service_cancel.py @@ -0,0 +1,52 @@ +"""ScrapeService cancels queued executor work on outer timeout.""" + +from __future__ import annotations + +import unittest +from concurrent.futures import ThreadPoolExecutor +from unittest.mock import MagicMock, patch + +from app.config import get_settings +from app.domain.scrape_service import ScrapeOutcome, ScrapeService +from app.engine import ScraperEngine +from app.schemas.enums import ErrorCategory, TimeoutPhase +from app.schemas.response import ScrapeError +from tests.support.factories import scrape_request + + +class ScrapeServiceCancelTests(unittest.IsolatedAsyncioTestCase): + async def test_timeout_cancels_executor_future(self) -> None: + settings = get_settings() + engine = ScraperEngine(settings=settings) + service = ScrapeService( + settings=settings, + engine=engine, + executor=ThreadPoolExecutor(max_workers=1), + ) + future = MagicMock() + future.cancel = MagicMock(return_value=True) + + async def boom(awaitable: object, timeout: float | None = None) -> None: + del awaitable, timeout + raise TimeoutError + + with ( + patch("asyncio.get_running_loop") as mock_loop, + patch("asyncio.wait_for", side_effect=boom), + ): + mock_loop.return_value.run_in_executor = MagicMock(return_value=future) + outcome = await service.process(scrape_request()) + + future.cancel.assert_called_once() + self.assertIsInstance(outcome, ScrapeOutcome) + self.assertEqual(outcome.status_code, 504) + self.assertIsInstance(outcome.body, ScrapeError) + assert isinstance(outcome.body, ScrapeError) + self.assertEqual(outcome.body.error_category, ErrorCategory.TIMEOUT) + self.assertEqual(outcome.body.diagnostics.timeout_phase, TimeoutPhase.QUEUE) + self.assertEqual(outcome.body.error, "Scraper at capacity; retry shortly") + service.executor.shutdown(wait=False, cancel_futures=True) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/domain/test_timeout_error.py b/tests/domain/test_timeout_error.py index 3f67b40..961d1fe 100644 --- a/tests/domain/test_timeout_error.py +++ b/tests/domain/test_timeout_error.py @@ -6,6 +6,7 @@ import unittest from app.domain.scrape_service import ScrapeService +from app.engine.envelope import TIMEOUT_ERROR_BY_PHASE from app.infra.scrape_progress import ScrapeProgress from app.schemas.enums import ExecutionTier, NavigationMode, TimeoutPhase @@ -22,11 +23,24 @@ def test_queue_phase_keeps_zero_attempts(self): timeout_seconds=45, ) self.assertEqual(result.error_category.value, "timeout") - self.assertIn("phase=queue", result.error) + self.assertEqual(result.error, TIMEOUT_ERROR_BY_PHASE[TimeoutPhase.QUEUE]) self.assertEqual(result.diagnostics.timeout_phase, TimeoutPhase.QUEUE) self.assertEqual(result.diagnostics.attempts, 0) self.assertIsNone(result.diagnostics.strategy_used) + def test_boot_phase_message(self): + progress = ScrapeProgress() + progress.mark(TimeoutPhase.BOOT, execution_tier=ExecutionTier.BROWSER_DRIVER) + result = ScrapeService.build_timeout_error( + _URL, + request_id="req-boot", + started_monotonic=time.monotonic(), + progress=progress, + timeout_seconds=45, + ) + self.assertEqual(result.error, TIMEOUT_ERROR_BY_PHASE[TimeoutPhase.BOOT]) + self.assertEqual(result.diagnostics.timeout_phase, TimeoutPhase.BOOT) + def test_work_phase_preserves_attempts_and_strategy(self): progress = ScrapeProgress() progress.mark( @@ -47,7 +61,7 @@ def test_work_phase_preserves_attempts_and_strategy(self): self.assertEqual(diagnostics.attempts, 2) self.assertEqual(diagnostics.strategy_used, NavigationMode.GOOGLE_GET) self.assertEqual(diagnostics.execution_tier, ExecutionTier.BROWSER_DRIVER) - self.assertIn("phase=work", result.error) + self.assertEqual(result.error, TIMEOUT_ERROR_BY_PHASE[TimeoutPhase.WORK]) self.assertGreaterEqual(diagnostics.render_ms, 0) diff --git a/tests/engine/test_browser_tier_challenge.py b/tests/engine/test_browser_tier_challenge.py index 83cdf11..1dd27e2 100644 --- a/tests/engine/test_browser_tier_challenge.py +++ b/tests/engine/test_browser_tier_challenge.py @@ -153,3 +153,27 @@ def test_soft_challenge_retries_strategies_then_blocks(self) -> None: self.assertEqual(_SoftChallengeDriver.navigate_calls, 3) assert result.diagnostics.challenge is not None self.assertTrue(result.diagnostics.challenge.detected) + + def test_soft_challenge_does_not_retry_when_work_budget_low(self) -> None: + """Below soft-retry floor, unclean assessment is challenge_block immediately.""" + _SoftChallengeDriver.reset() + payload = scrape_request( + execution_mode=ExecutionMode.BROWSER, + navigation_mode=NavigationMode.AUTO, + max_retries=2, + ) + with tempfile.TemporaryDirectory() as tmp: + engine = ScraperEngine(settings=get_settings(), runtime_root=Path(tmp)) + with ( + patch("botasaurus.browser.Driver", _SoftChallengeDriver), + patch( + "app.engine.browser_tier.remaining_work_seconds", + return_value=4, + ), + ): + result = engine.execute(payload, request_id="req-soft-low-budget") + self.assertIsInstance(result, ScrapeError) + assert isinstance(result, ScrapeError) + self.assertEqual(result.error_category, ErrorCategory.CHALLENGE_BLOCK) + self.assertEqual(result.diagnostics.attempts, 1) + self.assertEqual(_SoftChallengeDriver.navigate_calls, 1) diff --git a/tests/engine/test_budget.py b/tests/engine/test_budget.py index 60b8b63..a5fcf7f 100644 --- a/tests/engine/test_budget.py +++ b/tests/engine/test_budget.py @@ -29,13 +29,13 @@ def test_remaining_total_counts_down_from_scrape_timeout(self): self.assertLessEqual(fresh, self.settings.scrape_timeout_seconds) self.assertGreaterEqual(fresh, self.settings.scrape_timeout_seconds - 1) - def test_remaining_total_floors_at_one_when_exhausted(self): + def test_remaining_total_returns_zero_when_exhausted(self): exhausted = self.now - (self.settings.scrape_timeout_seconds + 60) - self.assertEqual(remaining_total_seconds(self.settings, exhausted), 1) + self.assertEqual(remaining_total_seconds(self.settings, exhausted), 0) - def test_remaining_work_floors_at_one_when_exhausted(self): + def test_remaining_work_returns_zero_when_exhausted(self): exhausted = self.now - (self.settings.scrape_work_timeout_seconds + 60) - self.assertEqual(remaining_work_seconds(self.settings, exhausted), 1) + self.assertEqual(remaining_work_seconds(self.settings, exhausted), 0) def test_browser_step_budget_takes_the_tighter_constraint(self): # Total budget nearly burnt, work budget fresh: total wins. @@ -48,10 +48,10 @@ def test_browser_step_budget_takes_the_tighter_constraint(self): self.assertLessEqual(fresh, self.settings.scrape_work_timeout_seconds) self.assertGreaterEqual(fresh, 1) - def test_browser_step_budget_never_returns_zero_or_negative(self): + def test_browser_step_budget_returns_zero_when_past_deadline(self): long_ago = self.now - 10_000 self.assertEqual( - browser_step_budget_seconds(self.settings, long_ago, long_ago), 1 + browser_step_budget_seconds(self.settings, long_ago, long_ago), 0 ) diff --git a/tests/engine/test_scraper_engine.py b/tests/engine/test_scraper_engine.py index 5bf16a0..4aa9fff 100644 --- a/tests/engine/test_scraper_engine.py +++ b/tests/engine/test_scraper_engine.py @@ -16,7 +16,6 @@ ErrorCategory, ExecutionTier, NavigationMode, - TimeoutPhase, ) from app.schemas.response import ( ScrapeDiagnostics, @@ -46,12 +45,14 @@ def get(self, *_args: object, **kwargs: Any) -> None: ) monotonic_values = [ - 1000.0, # execute started + 1000.0, # execute started (orchestrator) + 1000.0, # pre-boot remaining_total check 1000.0, # boot_started 1020.0, # boot_ms end 1020.0, # browser ready after boot - 1020.0, # remaining total - 1020.0, # remaining work + 1020.0, # remaining total (step budget) + 1020.0, # remaining work (step budget) + 1020.0, # remaining work (soft-retry gate) 1020.0, # render_ms ] @@ -242,7 +243,7 @@ def test_browser_driver_constructor_failure_returns_navigation_error(self): self.assertIsInstance(result, ScrapeError) assert isinstance(result, ScrapeError) self.assertEqual(result.error_category, ErrorCategory.NAVIGATION_ERROR) - self.assertEqual(result.diagnostics.timeout_phase, TimeoutPhase.BOOT) + self.assertIsNone(result.diagnostics.timeout_phase) self.assertEqual( result.diagnostics.execution_tier, ExecutionTier.BROWSER_DRIVER ) @@ -426,6 +427,36 @@ def test_request_tier_blocked_status_escalates_to_browser(self): result.headers["content-type"], "text/html; charset=utf-8" ) + def test_request_tier_skip_escalate_returns_challenge_not_timeout(self): + from app.engine.budget import MIN_ESCALATE_REMAINING_SECONDS + from tests.support.fakes import FakeHttpResponse, FakeRequest + + payload = scrape_request() + FakeRequest.response = FakeHttpResponse( + text="Just a moment...", + status_code=403, + headers={"content-type": "text/html"}, + url="https://example.com/", + ) + with tempfile.TemporaryDirectory() as tmp: + engine = ScraperEngine(settings=get_settings(), runtime_root=Path(tmp)) + with ( + patch("botasaurus.request.Request", FakeRequest), + patch( + "app.engine.request_tier.remaining_total_seconds", + side_effect=[30, MIN_ESCALATE_REMAINING_SECONDS - 1], + ), + patch("botasaurus.browser.Driver") as mock_driver, + ): + result = engine.execute(payload) + + mock_driver.assert_not_called() + self.assertIsInstance(result, ScrapeError) + assert isinstance(result, ScrapeError) + self.assertEqual(result.error_category, ErrorCategory.CHALLENGE_BLOCK) + self.assertIsNone(result.diagnostics.timeout_phase) + self.assertEqual(result.diagnostics.execution_tier, ExecutionTier.HTTP_REQUEST) + def test_html_response_sets_utf8_content_type_and_normalizes_body(self): from tests.support.fakes import FakeHttpResponse, FakeRequest diff --git a/tests/infra/test_challenge_detector.py b/tests/infra/test_challenge_detector.py index be5bb02..4d28817 100644 --- a/tests/infra/test_challenge_detector.py +++ b/tests/infra/test_challenge_detector.py @@ -27,13 +27,18 @@ def test_clean_response(self): def test_soft_challenge_may_retry_strategies(self): soft = ChallengeDetector.detect("Just a moment...", 200) - self.assertTrue(soft.may_retry_strategies(has_more=True)) - self.assertFalse(soft.may_retry_strategies(has_more=False)) + self.assertTrue(soft.may_retry_strategies(has_more=True, remaining_seconds=30)) + self.assertFalse( + soft.may_retry_strategies(has_more=False, remaining_seconds=30) + ) + self.assertFalse(soft.may_retry_strategies(has_more=True, remaining_seconds=4)) def test_hard_block_does_not_retry_strategies(self): hard = ChallengeDetector.detect("Forbidden", 403) - self.assertFalse(hard.may_retry_strategies(has_more=True)) - self.assertFalse(hard.may_retry_strategies(has_more=False)) + self.assertFalse(hard.may_retry_strategies(has_more=True, remaining_seconds=30)) + self.assertFalse( + hard.may_retry_strategies(has_more=False, remaining_seconds=30) + ) def test_driver_bot_detection_integration(self): mock_driver = MagicMock() From ab0c7149652e8af14c1a48d55967ac38efe5982a Mon Sep 17 00:00:00 2001 From: Gil Desmarais Date: Sat, 29 Aug 2026 00:32:19 +0200 Subject: [PATCH 2/4] fix(engine): probe challenges mid-wait in readiness chunks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Chunk selector waits to ≤2s and run ChallengeDetector between chunks so interstitials fail closed as challenge_block before burning the full wait budget. Boyscout: call_if_available in detector, single unclean surfacing helper, drop dead inspect params. --- app/engine/browser_tier.py | 49 +++++++++++------- app/engine/strategies.py | 55 ++++++++++++++++----- app/infra/detector.py | 27 +++++----- tests/engine/test_browser_tier_challenge.py | 31 ++++++++++++ tests/engine/test_scraper_engine.py | 33 ++++++++++++- 5 files changed, 151 insertions(+), 44 deletions(-) diff --git a/app/engine/browser_tier.py b/app/engine/browser_tier.py index 8ce7d70..bfa60a9 100644 --- a/app/engine/browser_tier.py +++ b/app/engine/browser_tier.py @@ -67,9 +67,7 @@ def settle_page_state( return html, meta, assessment, xhr_responses -def _inspect_assessment( - driver: DriverProtocol, _target_url: str -) -> ChallengeAssessment | None: +def _inspect_assessment(driver: DriverProtocol) -> ChallengeAssessment | None: """Best-effort challenge assessment after a nav/timeout exception. Lean path only: html + passive request status + driver signals via @@ -114,7 +112,7 @@ def _challenge_block_error( ) -def _unclean_retry_or_block( +def _surface_unclean( target_url: str, *, request_id: str, @@ -126,7 +124,16 @@ def _unclean_retry_or_block( remaining_work: int, collector: XhrCollector, ) -> ScrapeError | None: - """Retry soft challenges; otherwise return challenge_block. None ⇒ continue.""" + """Log and apply unclean assessment. None ⇒ soft-retry / continue.""" + logger.warning( + "scrape_challenge_detected request_id=%s host=%s strategy=%s " + "attempt=%d marker=%s", + request_id, + urlparse(target_url).hostname, + strategy.value, + attempts, + assessment.detected_marker, + ) if assessment.may_retry_strategies( has_more=has_more_strategies, remaining_seconds=remaining_work, @@ -341,11 +348,26 @@ def run_browser_tier( remaining_work = remaining_work_seconds(settings, browser_ready_monotonic) try: navigate(driver, target_url, strategy, step_budget) - wait_for_readiness( + mid_wait = wait_for_readiness( driver, selector=payload.wait_for_selector, timeout_seconds=min(payload.wait_timeout_seconds, step_budget), ) + if mid_wait is not None: + blocked = _surface_unclean( + target_url, + request_id=request_id, + attempts=attempts, + strategy=strategy, + started_monotonic=started_monotonic, + assessment=mid_wait, + has_more_strategies=has_more, + remaining_work=remaining_work, + collector=collector, + ) + if blocked is None: + continue + return blocked if payload.scroll: apply_scrolling(driver) @@ -355,16 +377,7 @@ def run_browser_tier( ) if not assessment.is_clean: - logger.warning( - "scrape_challenge_detected request_id=%s host=%s strategy=%s " - "attempt=%d marker=%s", - request_id, - urlparse(target_url).hostname, - strategy.value, - attempt_index, - assessment.detected_marker, - ) - blocked = _unclean_retry_or_block( + blocked = _surface_unclean( target_url, request_id=request_id, attempts=attempts, @@ -406,9 +419,9 @@ def run_browser_tier( str(exc), ) # Prefer knowable challenge over timeout when the page is inspectable. - assessment = _inspect_assessment(driver, target_url) + assessment = _inspect_assessment(driver) if assessment is not None and not assessment.is_clean: - blocked = _unclean_retry_or_block( + blocked = _surface_unclean( target_url, request_id=request_id, attempts=attempts, diff --git a/app/engine/strategies.py b/app/engine/strategies.py index 425a782..7577fef 100644 --- a/app/engine/strategies.py +++ b/app/engine/strategies.py @@ -12,6 +12,7 @@ resolve_callable, resolve_cdp_tab, ) +from app.infra.detector import ChallengeAssessment, ChallengeDetector from app.infra.xhr_collector import XhrCollector from app.logging_config import get_logger from app.schemas.enums import NavigationMode @@ -21,6 +22,7 @@ logger = get_logger() _CAPABILITY_MISS = object() +_READINESS_CHUNK_SECONDS = 2 TRACKER_URL_PATTERNS: list[str] = [ "*google-analytics.com*", @@ -111,22 +113,53 @@ def configure_driver( ) +def _mid_wait_challenge(driver: DriverProtocol) -> ChallengeAssessment | None: + """Best-effort challenge probe during readiness wait. None if clean or unreadable.""" + try: + html = driver.page_html or "" + except Exception: + return None + assessment = ChallengeDetector.detect(html, driver=driver) + if assessment.is_clean: + return None + return assessment + + def wait_for_readiness( driver: DriverProtocol, *, selector: str | None, timeout_seconds: int, -) -> None: - if selector: - driver.wait_for_element(selector, wait=timeout_seconds) - return - - if ( - call_if_available(driver, "sleep_random", 0.5, 1.2, default=_CAPABILITY_MISS) - is not _CAPABILITY_MISS - ): - return - driver.sleep(1) +) -> ChallengeAssessment | None: + """Wait for selector / settle; return unclean assessment if challenge appears mid-wait. + + Selector waits run in ≤2s chunks so a challenge interstitial can fail closed + before the full wait budget burns. Non-selector settles stay short and probe once. + """ + if not selector: + if ( + call_if_available( + driver, "sleep_random", 0.5, 1.2, default=_CAPABILITY_MISS + ) + is _CAPABILITY_MISS + ): + driver.sleep(1) + return _mid_wait_challenge(driver) + + remaining = max(0, int(timeout_seconds)) + while remaining > 0: + chunk = min(_READINESS_CHUNK_SECONDS, remaining) + try: + driver.wait_for_element(selector, wait=chunk) + return None + except Exception: + assessment = _mid_wait_challenge(driver) + if assessment is not None: + return assessment + remaining -= chunk + + driver.wait_for_element(selector, wait=0) + return None def apply_scrolling(driver: DriverProtocol) -> None: diff --git a/app/infra/detector.py b/app/infra/detector.py index dd4c3bb..d77dc9d 100644 --- a/app/infra/detector.py +++ b/app/infra/detector.py @@ -3,7 +3,9 @@ from __future__ import annotations from dataclasses import dataclass +from typing import cast +from app.engine.driver_capabilities import DriverProtocol, call_if_available from app.schemas.response import ChallengeSignal _CHALLENGE_MARKERS: tuple[str, ...] = ( @@ -29,6 +31,12 @@ # Soft strategy retries need this much remaining work budget (seconds). _MIN_SOFT_RETRY_REMAINING_SECONDS = 5 +_DRIVER_SIGNAL_METHODS: tuple[tuple[str, str], ...] = ( + ("is_bot_detected", "botasaurus_driver_bot_detected"), + ("is_in_challenge", "botasaurus_driver_challenge"), + ("is_blocked", "botasaurus_driver_blocked"), +) + @dataclass(frozen=True, slots=True) class ChallengeAssessment: @@ -74,20 +82,11 @@ def detect( # 1. Driver-level anti-bot signal inspection if driver is not None: - for method_name, marker_label in ( - ("is_bot_detected", "botasaurus_driver_bot_detected"), - ("is_in_challenge", "botasaurus_driver_challenge"), - ("is_blocked", "botasaurus_driver_blocked"), - ): - check_fn = getattr(driver, method_name, None) - if callable(check_fn): - try: - if check_fn(): - matched_marker = marker_label - break - except Exception: - # Best-effort driver bot detection check - pass + typed = cast(DriverProtocol, driver) + for method_name, marker_label in _DRIVER_SIGNAL_METHODS: + if call_if_available(typed, method_name, default=False): + matched_marker = marker_label + break # 2. HTML text markers (only inspect if driver check did not match) if matched_marker is None and html: diff --git a/tests/engine/test_browser_tier_challenge.py b/tests/engine/test_browser_tier_challenge.py index 1dd27e2..cf86350 100644 --- a/tests/engine/test_browser_tier_challenge.py +++ b/tests/engine/test_browser_tier_challenge.py @@ -177,3 +177,34 @@ def test_soft_challenge_does_not_retry_when_work_budget_low(self) -> None: self.assertEqual(result.error_category, ErrorCategory.CHALLENGE_BLOCK) self.assertEqual(result.diagnostics.attempts, 1) self.assertEqual(_SoftChallengeDriver.navigate_calls, 1) + + def test_mid_wait_challenge_returns_challenge_block(self) -> None: + class _MidWaitChallengeDriver(_ScenarioDriver): + navigate_calls: ClassVar[int] = 0 + page_body = "Just a moment..." + wait_calls: ClassVar[int] = 0 + + def wait_for_element(self, *_args: object, **_kwargs: Any) -> None: + type(self).wait_calls += 1 + raise TimeoutError("element not found") + + _MidWaitChallengeDriver.reset() + _MidWaitChallengeDriver.wait_calls = 0 + payload = scrape_request( + execution_mode=ExecutionMode.BROWSER, + navigation_mode=NavigationMode.GET, + max_retries=0, + wait_for_selector="#content", + wait_timeout_seconds=4, + ) + with tempfile.TemporaryDirectory() as tmp: + engine = ScraperEngine(settings=get_settings(), runtime_root=Path(tmp)) + with patch("botasaurus.browser.Driver", _MidWaitChallengeDriver): + result = engine.execute(payload, request_id="req-mid-wait") + self.assertIsInstance(result, ScrapeError) + assert isinstance(result, ScrapeError) + self.assertEqual(result.error_category, ErrorCategory.CHALLENGE_BLOCK) + self.assertIsNone(result.diagnostics.timeout_phase) + # First chunk probes challenge and fails closed — no full 4s burn. + self.assertEqual(_MidWaitChallengeDriver.wait_calls, 1) + self.assertEqual(_MidWaitChallengeDriver.navigate_calls, 1) diff --git a/tests/engine/test_scraper_engine.py b/tests/engine/test_scraper_engine.py index 4aa9fff..af4b827 100644 --- a/tests/engine/test_scraper_engine.py +++ b/tests/engine/test_scraper_engine.py @@ -172,8 +172,39 @@ def test_scrape_envelope_constructors(self): def test_wait_for_readiness_uses_sleep_random_when_available(self): mock_driver = MagicMock() mock_driver.sleep_random = MagicMock() - wait_for_readiness(mock_driver, selector=None, timeout_seconds=10) + mock_driver.page_html = "ok" + mock_driver.is_bot_detected.return_value = False + mock_driver.is_in_challenge.return_value = False + mock_driver.is_blocked.return_value = False + result = wait_for_readiness(mock_driver, selector=None, timeout_seconds=10) mock_driver.sleep_random.assert_called_once_with(0.5, 1.2) + self.assertIsNone(result) + + def test_wait_for_readiness_chunks_selector_and_surfaces_challenge(self): + from typing import cast + + from app.engine.driver_capabilities import DriverProtocol + from app.engine.strategies import wait_for_readiness as readiness + + class _ChunkDriver(FakeDriver): + wait_calls = 0 + + def wait_for_element(self, *_args: object, **_kwargs: Any) -> None: + type(self).wait_calls += 1 + raise TimeoutError("missing") + + def __init__(self, *args: object, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self.page_html = "Just a moment..." + + driver = _ChunkDriver() + assessment = readiness( + cast(DriverProtocol, driver), selector="#x", timeout_seconds=6 + ) + self.assertIsNotNone(assessment) + assert assessment is not None + self.assertFalse(assessment.is_clean) + self.assertEqual(_ChunkDriver.wait_calls, 1) def test_execute_honors_submission_deadline_after_queue_wait(self): settings = get_settings() From 541a3e3b4d7313e665dba40404c8385a893497b6 Mon Sep 17 00:00:00 2001 From: Gil Desmarais Date: Sat, 29 Aug 2026 00:33:22 +0200 Subject: [PATCH 3/4] docs: add timeout_phase ops triage table Map queue/boot/work and challenge_block to capacity vs product signals, and note 504 Future.cancel behavior. --- README.md | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/README.md b/README.md index 9a34f3b..2612f83 100644 --- a/README.md +++ b/README.md @@ -269,6 +269,17 @@ Status codes: - `502`: scrape execution failure/challenge block. - `504`: scrape timed out. +### Timeout / challenge triage + +| Signal | Ops meaning | +| :--- | :--- | +| `timeout_phase=queue` | Capacity/workers — threadpool wait burned the budget | +| `timeout_phase=boot` | RAM/shm/prewarm canary — browser/driver failed to start in time | +| `timeout_phase=work` | Slow or hostile target — navigate/wait/scroll exceeded budget | +| `challenge_block` | Product signal — anti-bot surface (502), not a timeout | + +On outer **504**, queued Futures are cancelled (work never starts). In-flight scrapes stop at the next deadline check or step timeout. + ## Runtime Flow And Invariants `POST /scrape` executes this path: From 06d058ff8d7b92f34f7d7cf99115c6ebb58626fd Mon Sep 17 00:00:00 2001 From: Gil Desmarais Date: Sat, 29 Aug 2026 00:52:49 +0200 Subject: [PATCH 4/4] fix(engine): clear timeout_phase on storage ENOSPC errors Storage failures are NAVIGATION_ERROR, not timeouts; keep timeout_phase reserved for real timeout outcomes so ops/Sentry tags stay honest. --- app/engine/browser_tier.py | 2 +- tests/engine/test_scraper_engine.py | 4 +--- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/app/engine/browser_tier.py b/app/engine/browser_tier.py index bfa60a9..c325d56 100644 --- a/app/engine/browser_tier.py +++ b/app/engine/browser_tier.py @@ -169,7 +169,7 @@ def _boot_storage_error( render_ms=elapsed_ms(started_monotonic), error_category=ErrorCategory.NAVIGATION_ERROR, execution_tier=ExecutionTier.BROWSER_DRIVER, - timeout_phase=TimeoutPhase.BOOT, + timeout_phase=None, ) diff --git a/tests/engine/test_scraper_engine.py b/tests/engine/test_scraper_engine.py index af4b827..32ffc39 100644 --- a/tests/engine/test_scraper_engine.py +++ b/tests/engine/test_scraper_engine.py @@ -328,9 +328,7 @@ def boom_mkdir( assert isinstance(result, ScrapeError) self.assertEqual(result.error_category, ErrorCategory.NAVIGATION_ERROR) self.assertIn("runtime storage full", result.error) - self.assertIsNotNone(result.diagnostics.timeout_phase) - assert result.diagnostics.timeout_phase is not None - self.assertEqual(result.diagnostics.timeout_phase.value, "boot") + self.assertIsNone(result.diagnostics.timeout_phase) self.assertEqual(list(runtime_root.iterdir()), []) def test_prune_orphan_runtime_dirs_before_new_request(self):