fix(scrape): cancel queued work and fail closed on budget

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:

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"],

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,

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,
@@ -66,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
@@ -113,7 +112,7 @@ def _challenge_block_error(
     )
 
 
-def _unclean_retry_or_block(
+def _surface_unclean(
     target_url: str,
     *,
     request_id: str,
@@ -122,10 +121,23 @@ 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):
+    """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,
+    ):
         collector.reset()
         return None
     return _challenge_block_error(
@@ -157,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,
     )
 
 
@@ -172,9 +184,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 +200,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,16 +334,40 @@ 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(
+            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)
@@ -315,23 +377,15 @@ 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,
                     strategy=strategy,
                     started_monotonic=started_monotonic,
                     assessment=assessment,
                     has_more_strategies=has_more,
+                    remaining_work=remaining_work,
                     collector=collector,
                 )
                 if blocked is None:
@@ -365,16 +419,19 @@ 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,
                     strategy=strategy,
                     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:

app/engine/budget.py

@@ -6,21 +6,25 @@
 
 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)
 
 
 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)

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():

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,