Skip to content

feat(engine): WorkLease reclaim and fail-closed challenge_block #53

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
gildesmarais merged 5 commits into from
Aug 30, 2026
Merged

feat(engine): WorkLease reclaim and fail-closed challenge_block #53

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
5ce4d78
feat(engine): WorkLease owns admit, deadline reclaim, and timeout phase
gildesmarais Aug 30, 2026
e93eafb
fix(engine): fail closed unclean/hung surfaces as challenge_block
gildesmarais Aug 30, 2026
dcb4ac7
test(challenge): share interstitial fixture corpus with gem
gildesmarais Aug 30, 2026
1c318d8
fix(session): keep reclaim close effective across cold-boot race
gildesmarais Aug 30, 2026
837bf85
fix(lease): abort before Chromium and fail-closed mid-wait unreadables
gildesmarais Aug 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 24 additions & 15 deletions AGENTS.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,10 +21,11 @@ app/
openapi_examples.py # OpenAPI examples built from Pydantic model instances
routes/ # thin HTTP handlers (health, scrape)
domain/
scrape_service.py # request-id resolution, URL guardrails, threadpool execution, status mapping
scrape_service.py # request-id resolution, URL guardrails, WorkLease execution, status mapping
engine/
orchestrator.py # ScraperEngine.execute
session.py # ScrapeSession lifecycle
session.py # ScrapeSession lifecycle + lease reclaim hook
work_lease.py # WorkLease admit/deadline/reclaim + phase snapshot; HostConcurrencyGate
budget.py # wall-clock budget math shared across tiers (elapsed_ms, step budgets)
request_tier.py # HTTP/curl_cffi path
browser_tier.py # Chromium path
Expand All @@ -36,23 +37,27 @@ app/
enums.py # ExecutionMode, NavigationMode, ErrorCategory, ...
request.py # ScrapeRequest and validators
response.py # ScrapeSuccess, ScrapeError, HealthResponse, ...
infra/ # telemetry, progress, metadata, xhr, runtime cleanup, sentry
infra/ # telemetry, metadata, xhr, runtime cleanup, sentry, challenge detector
# (phase snapshot lives on WorkLease — not a separate progress module)
security/ # UrlGuard SSRF guardrails
scripts/
bench_scrape.py # TestClient wall-time bench for POST /scrape (request tier)
tests/
api/ # HTTP contract, request schema, 504 timeout envelope tests
domain/ # ScrapeService unit tests (timeout error mapping)
engine/ # ScraperEngine units, isolation regressions, timeout progress
infra/ # challenge, metadata, xhr, progress, sentry, telemetry, request-id, cleanup
security/ # UrlGuard tests
support/
http.py # test_client() context manager + dependency_overrides helper
fakes.py # shared FakeDriver, FakeRequest, fake_request_cls, ...
factories.py # scrape_request(), example_url()
test_bench_regression.py # lightweight guard that bench script completes (root: guards scripts/)
tests/
api/ # HTTP contract, request schema, 504 timeout envelope tests
domain/ # ScrapeService unit tests (timeout error mapping)
engine/ # ScraperEngine units, isolation regressions, timeout progress
fixtures/challenge/ # shared interstitial HTML corpus (gem BlockedSurface loads sibling path)
infra/ # challenge, metadata, xhr, sentry, telemetry, request-id, cleanup
security/ # UrlGuard tests
support/
http.py # test_client() context manager + dependency_overrides helper
fakes.py # shared FakeDriver, FakeRequest, fake_request_cls, ...
factories.py # scrape_request(), example_url()
test_bench_regression.py # lightweight guard that bench script completes (root: guards scripts/)
```

**Challenge corpus:** HTML under `tests/fixtures/challenge/` is the single fixture home for interstitial detection. Scrape-api `ChallengeDetector` markers and gem `BlockedSurface` signatures must both assert those files; do not reintroduce one-sided markers without a shared fixture.

Layer rules:

| Layer | May import | Must not import |
Expand Down Expand Up @@ -91,11 +96,15 @@ Conventions:
| --- | --- | --- |
| `ScraperEngine` | process (app.state) | shared |
| `ThreadPoolExecutor` | process (app.state) | shared, sized by `SCRAPE_MAX_WORKERS` |
| `HostConcurrencyGate` | process (ScrapeService) | per-host admit; sized by `SCRAPE_MAX_PER_HOST` |
| `WorkLease` | per request | owns admit → executor run → deadline reclaim (session `force_close`); phase snapshot for timeout envelope; `Future.cancel` is not Chromium reclaim |
| `WarmDriverPool` (opt-in) | process (engine.warm_pool) | single spare slot; refill on dedicated daemon thread — never the scrape executor |
| `_active_request_ids` | in-process memory | shared; collision guard |
| runtime dir `/tmp/scrape/<request_id>` | per request | isolated; deleted in `finally` |
| browser profile | per request (or adopted spare-*) | isolated; no reuse across requests; warm spare dies with the adopting request |
| Botasaurus Driver | may start before assignment | usage stays ≤1 request; closed in session `__exit__`; never returned to the pool |
| Botasaurus Driver | may start before assignment | usage stays ≤1 request; closed in session `__exit__` / lease reclaim; never returned to the pool |

**Timeout ownership:** `WorkLease` is the single owner of outer deadline, host admit, phase snapshot, and Chromium reclaim. Tiers mark phase on the lease; `WorkLease.timeout_error` is the only outer-timeout envelope builder. Session exit and orphan prune remain safety nets under lease reclaim — not a second reclaim story.

Multi-worker uvicorn breaks in-process collision detection unless request ids are sticky to a worker. Default to single-worker for isolation semantics.

Expand Down
3 changes: 3 additions & 0 deletions app/config.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,9 @@ class Settings(BaseSettings):
default=30, validation_alias="SCRAPE_WORK_TIMEOUT_SECONDS"
)
scrape_max_workers: int = Field(default=4, validation_alias="SCRAPE_MAX_WORKERS")
scrape_max_per_host: int = Field(
default=2, ge=1, validation_alias="SCRAPE_MAX_PER_HOST"
)
scrape_runtime_min_free_bytes: int = Field(
default=256 * 1024 * 1024,
validation_alias="SCRAPE_RUNTIME_MIN_FREE_BYTES",
Expand Down
69 changes: 17 additions & 52 deletions app/domain/scrape_service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,20 +2,17 @@

from __future__ import annotations

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from functools import partial

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.engine.work_lease import HostConcurrencyGate, WorkLease
from app.exceptions import RequestIdCollisionError
from app.infra.ops_telemetry import emit_terminal_telemetry
from app.infra.request_id import resolve_request_id
from app.infra.scrape_progress import ScrapeProgress
from app.logging_config import get_logger
from app.schemas.enums import ErrorCategory, TimeoutPhase
from app.schemas.request import ScrapeRequest
Expand All @@ -37,18 +34,20 @@ class ScrapeOutcome:


class ScrapeService:
"""Owns request-id resolution, URL guardrails, threadpool execution, status mapping, and telemetry."""
"""Owns request-id resolution, URL guardrails, WorkLease execution, status mapping."""

def __init__(
self,
*,
settings: Settings,
engine: ScraperEngine,
executor: ThreadPoolExecutor,
host_gate: HostConcurrencyGate | None = None,
) -> None:
self.settings = settings
self.engine = engine
self.executor = executor
self.host_gate = host_gate or HostConcurrencyGate(settings.scrape_max_per_host)

async def process(
self,
Expand Down Expand Up @@ -108,65 +107,33 @@ def _validation_outcome(
status_code=validation.status_code,
)

@staticmethod
def build_timeout_error(
url: str,
*,
request_id: str,
started_monotonic: float,
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=TIMEOUT_ERROR_BY_PHASE[phase],
error_category=ErrorCategory.TIMEOUT,
diagnostics=ScrapeDiagnostics(
request_id=request_id,
attempts=snap.attempts,
strategy_used=snap.strategy_used,
render_ms=render_ms,
execution_tier=snap.execution_tier,
timeout_phase=phase,
),
)

async def _run(
self,
payload: ScrapeRequest,
*,
request_id: str,
) -> ScrapeOutcome:
target_url = str(payload.url)
host = payload.url.host
host = payload.url.host or ""
lease = WorkLease(
settings=self.settings,
executor=self.executor,
host_gate=self.host_gate,
)
started_monotonic = time.monotonic()
deadline_monotonic = started_monotonic + self.settings.scrape_timeout_seconds
progress = ScrapeProgress()

try:
loop = asyncio.get_running_loop()
future = loop.run_in_executor(
self.executor,
partial(
result = await lease.run(
host=host,
work=partial(
self.engine.execute,
payload,
deadline_monotonic,
request_id=request_id,
progress=progress,
lease=lease,
),
)
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,
Expand All @@ -181,12 +148,10 @@ async def _run(
emit_terminal_telemetry(collision_result, http_status=502)
return ScrapeOutcome(body=collision_result, status_code=502)
except TimeoutError:
timeout_result = self.build_timeout_error(
timeout_result = lease.timeout_error(
target_url,
request_id=request_id,
started_monotonic=started_monotonic,
progress=progress,
timeout_seconds=self.settings.scrape_timeout_seconds,
)
phase = timeout_result.diagnostics.timeout_phase or TimeoutPhase.QUEUE
logger.warning(
Expand All @@ -200,7 +165,7 @@ async def _run(
emit_terminal_telemetry(
timeout_result,
http_status=504,
warm_hit=progress.snapshot().warm_hit,
warm_hit=lease.snapshot().warm_hit,
)
return ScrapeOutcome(body=timeout_result, status_code=504)

Expand All @@ -209,7 +174,7 @@ async def _run(
emit_terminal_telemetry(
result,
http_status=status_code,
warm_hit=progress.snapshot().warm_hit,
warm_hit=lease.snapshot().warm_hit,
)
logger.info(
"scrape_complete request_id=%s host=%s mode=%s tier=%s attempts=%s status=%d error_category=%s",
Expand Down
Loading