From 6fbdfb428f495d2efbe728a4d32c59bf610be757 Mon Sep 17 00:00:00 2001 From: ChechiDev Date: Sun, 2 Aug 2026 14:16:37 +0200 Subject: [PATCH 1/2] feat(player-info): integrate buffered mode execution gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add RateLimitGate (infrastructure/browser/gate.py): cooldown+probe recovery, idempotent signal_rate_limit(), cancel_recovery() for dead-engine teardown paths - Add release_to_pending() to PlayerInfoQueueRepository: IN_PROGRESS → PENDING without retry penalty (gate-after-claim race handler) - Wire gate into _run_buffered_loop: wait_if_closed() before navigation, gate-after-claim race release, signal on RateLimitError with probe_fn, cancel_recovery() in both engine teardown paths - Add 3 gate config fields to ScrapingSettings (cooldown, probe timeout, max probe attempts) - Add direct/buffered mode logging in main() - Add 16 unit tests for RateLimitGate, 10 for mode selection, 4 integration tests for release_to_pending() - Add manual soak procedure (docs/operations/player_info_buffered_mode_soak.md) --- config/settings.py | 4 + .../player_info_buffered_mode_soak.md | 183 +++++++++++ infrastructure/browser/gate.py | 180 +++++++++++ .../repositories/player_info_queue.py | 28 +- scripts/scrape_player_info.py | 76 +++++ .../test_player_info_queue_release.py | 119 +++++++ .../browser/test_dispatch_buffer.py | 15 +- .../browser/test_rate_limit_gate.py | 301 ++++++++++++++++++ .../infrastructure/player_info/__init__.py | 0 .../player_info/test_mode_selection.py | 104 ++++++ 10 files changed, 1000 insertions(+), 10 deletions(-) create mode 100644 docs/operations/player_info_buffered_mode_soak.md create mode 100644 infrastructure/browser/gate.py create mode 100644 tests/integration/test_player_info_queue_release.py create mode 100644 tests/unit/infrastructure/browser/test_rate_limit_gate.py create mode 100644 tests/unit/infrastructure/player_info/__init__.py create mode 100644 tests/unit/infrastructure/player_info/test_mode_selection.py diff --git a/config/settings.py b/config/settings.py index 6a49a40..2f93f1d 100644 --- a/config/settings.py +++ b/config/settings.py @@ -70,6 +70,10 @@ class ScrapingSettings(BaseModel): player_info_dispatch_buffer_size: int = Field(default=20, ge=1, le=500) player_info_dispatch_buffer_prefetch: int = Field(default=50, ge=1, le=500) player_info_dispatch_buffer_poll_interval: float = Field(default=5.0, ge=0.5) + # Rate-limit gate for buffered player_info mode + player_info_gate_cooldown_secs: float = Field(default=60.0, ge=5.0) + player_info_gate_probe_timeout_secs: float = Field(default=30.0, ge=5.0) + player_info_gate_max_probe_attempts: int = Field(default=3, ge=1, le=10) @model_validator(mode="after") def validate_warm_pool_ranges(self) -> "ScrapingSettings": diff --git a/docs/operations/player_info_buffered_mode_soak.md b/docs/operations/player_info_buffered_mode_soak.md new file mode 100644 index 0000000..9e5d4fc --- /dev/null +++ b/docs/operations/player_info_buffered_mode_soak.md @@ -0,0 +1,183 @@ +# Player Info Buffered Mode — Manual Soak Procedure + +**Status:** Disabled by default. Requires explicit feature flag to enable. + +**Feature flag:** `SCRAPING__PLAYER_INFO_DISPATCH_BUFFER_ENABLED=true` + +Do not enable in production without completing this soak procedure. + +--- + +## Prerequisites + +- A local or staging environment with a populated `scrape_queue` (player_info jobs in PENDING state). +- Direct database access to observe queue state during the soak. +- Ability to observe structured logs and process metrics (memory, file descriptors). +- The ability to terminate the process and re-run it with different settings. + +## What buffered mode changes + +When `SCRAPING__PLAYER_INFO_DISPATCH_BUFFER_ENABLED=true`: + +- A `CandidateProducer` fills an in-memory `BoundedCandidateBuffer` with PENDING job IDs (no claim, no lock). +- Workers consume job references from the buffer and claim the job from PostgreSQL at handoff. +- A `RateLimitGate` closes on `RateLimitError`, runs a bounded cooldown, and probes readiness before reopening. +- Direct PostgreSQL claiming by workers is disabled while buffered mode is active. +- The warm browser pool (`SCRAPING__PLAYER_INFO_WARM_POOL_ENABLED`) remains off by default and is separate from buffered mode. + +When `SCRAPING__PLAYER_INFO_DISPATCH_BUFFER_ENABLED=false` (default): + +- Direct claim mode is active. Workers call `claim_next()` directly from PostgreSQL. +- The buffer, producer, and rate-limit gate are not started. +- Existing behavior is fully preserved. + +--- + +## Rollback + +To return to direct mode at any time: + +1. Stop the running process (Ctrl+C or SIGTERM). +2. Remove or set `SCRAPING__PLAYER_INFO_DISPATCH_BUFFER_ENABLED=false`. +3. Restart the process normally. + +Do not manually edit the scrape_queue table unless using a documented recovery operation (`recover_all_stale`, `recover_failed`). + +--- + +## Soak Procedure + +### Stage 1 — 2 workers + +```bash +SCRAPING__PLAYER_INFO_DISPATCH_BUFFER_ENABLED=true \ + uv run python scripts/scrape_player_info.py --workers 2 +``` + +Run for at least 15 minutes. Record: + +- Throughput (jobs/minute from log output) +- Failure count and retry count (from queue: `SELECT status, count(*) FROM scrape_queue WHERE job_type='player_info' GROUP BY status`) +- Browser restart count (from logs: `WARNING Browser start failed`) +- Rate-limit gate events (from logs: `rate_limit_gate: CLOSED` / `REOPENED`) +- Cooldown events (from logs: `cooldown started` / `cooldown complete`) +- Readiness probe results (from logs: `probe passed` / `probe FAILED`) +- Memory (RSS): `ps -o rss= -p ` or equivalent +- Open file descriptors: `ls /proc//fd | wc -l` or `lsof -p | wc -l` +- DB queue state: pending/in_progress/done/failed row counts + +**Stop criteria at this stage:** +- Any repeated `rate_limit_gate: all N probes failed` message within 15 minutes → stop, investigate. +- Memory growing >500 MB above baseline without recovery → stop. +- File descriptor count growing continuously → stop. +- `FATAL: cannot record failure` error → stop immediately. +- Unexpected duplicate processing (same player_id inserted twice) → stop, investigate. + +### Stage 2 — 5 workers + +Only proceed if Stage 1 was stable for at least 15 minutes. + +```bash +SCRAPING__PLAYER_INFO_DISPATCH_BUFFER_ENABLED=true \ + uv run python scripts/scrape_player_info.py --workers 5 +``` + +Run for at least 20 minutes. Record the same metrics as Stage 1. + +**Stop criteria:** Same as Stage 1. + +### Stage 3 — 10 workers + +Only proceed if Stage 2 was stable. + +```bash +SCRAPING__PLAYER_INFO_DISPATCH_BUFFER_ENABLED=true \ + uv run python scripts/scrape_player_info.py --workers 10 +``` + +Run for at least 20 minutes. Record the same metrics. + +Watch especially for: +- Increased `claim_by_id` misses (logged as `dispatch: job N already claimed, skipping`): a moderate rate is expected and safe. +- DB lock contention: monitor `pg_stat_activity` for long-running lock waits. + +**Stop criteria:** Same as Stage 1. Also stop if DB lock waits exceed 10 seconds. + +### Stage 4 — 15 workers + +Only proceed if Stage 3 was stable. + +```bash +SCRAPING__PLAYER_INFO_DISPATCH_BUFFER_ENABLED=true \ + uv run python scripts/scrape_player_info.py --workers 15 +``` + +Run for at least 30 minutes. Record the same metrics. + +**Stop criteria:** Same as above. + +### Stage 5 — up to 25 workers (only if Stage 4 is stable) + +Only proceed if Stage 4 was stable for at least 30 minutes with no stop criteria triggered. + +```bash +SCRAPING__PLAYER_INFO_DISPATCH_BUFFER_ENABLED=true \ + uv run python scripts/scrape_player_info.py --workers 25 +``` + +Run for at least 30 minutes. Record the same metrics. + +--- + +## Stop Criteria (all stages) + +Stop immediately if ANY of these occur: + +- Repeated `rate_limit_gate: CLOSED` within a 5-minute window (3 or more closures) +- `rate_limit_gate: all N probes failed` → gate stuck closed +- Repeated readiness probe failures (`probe FAILED` ≥ 5 consecutive) +- Rising FAILED or STALE job count in the queue without recovery +- Memory (RSS) growing continuously for more than 10 minutes without recovery +- File descriptor count growing continuously for more than 10 minutes +- Browser restart storm: more than 5 restarts per worker per 10-minute window +- DB lock contention exceeding 10s +- Unexpected duplicate processing (same player written twice) +- `FATAL: cannot record failure` in any log + +## What to record at each stage + +For each stage, save a snapshot of: + +```sql +-- Queue state +SELECT status, count(*) FROM sch_fbref_infra.scrape_queue +WHERE job_type = 'player_info' GROUP BY status; + +-- Retry distribution +SELECT retry_count, count(*) FROM sch_fbref_infra.scrape_queue +WHERE job_type = 'player_info' GROUP BY retry_count ORDER BY retry_count; +``` + +And from logs: +- Count of `rate_limit_gate: CLOSED` events +- Count of `rate_limit_gate: REOPENED` events +- Count of `probe passed` vs `probe FAILED` events +- Count of `released job N to PENDING (gate closed)` events +- Count of `already claimed, skipping` events (claim race misses) + +--- + +## Security notes + +- Buffered mode does not log cookies, CDP tokens, WebSocket URLs, or browser session data. +- The candidate buffer contains only integer job IDs. +- Gate logs contain only reason codes and timing information. +- The readiness probe only reports pass/fail — no HTML or session content is logged. + +--- + +## Notes + +- Do not commit soak result logs to the repository. +- Do not scale beyond 25 workers — the UI and settings enforce a maximum of 25. +- If in doubt at any stage, roll back to direct mode and investigate before continuing. diff --git a/infrastructure/browser/gate.py b/infrastructure/browser/gate.py new file mode 100644 index 0000000..b9314b2 --- /dev/null +++ b/infrastructure/browser/gate.py @@ -0,0 +1,180 @@ +"""Rate-limit execution gate for buffered player_info mode. + +Shared gate object. Workers check is_open before navigating. +A single rate-limit signal closes the gate, runs a bounded cooldown, +optionally probes for readiness using an engine.warmup() call, then reopens. + +No DB dependency. No secrets logged. +""" +from __future__ import annotations + +import asyncio +import logging +import time +from collections.abc import Awaitable, Callable + +_log = logging.getLogger(__name__) + +_POLL_INTERVAL: float = 1.0 + + +class RateLimitGate: + """Shared execution gate for buffered player_info mode. + + Workers call wait_if_closed() before navigation. Any worker catching + a RateLimitError calls signal_rate_limit() with an optional probe_fn. + signal_rate_limit() is idempotent while recovery is in progress. + + After cooldown completes, the gate optionally probes readiness using + probe_fn (expected to call engine.warmup()). On probe success the gate + reopens. On probe failure it backs off and retries up to + max_probe_attempts times; if all fail the gate remains closed and + signal_rate_limit() must be called again by the next rate-limit event. + + Args: + cooldown_secs: Seconds to wait after a rate-limit signal before probing. + probe_timeout_secs: Per-probe timeout in seconds. + max_probe_attempts: Number of probe attempts before giving up. + sleep_fn: Async sleep callable; defaults to asyncio.sleep. Injected in tests. + """ + + def __init__( + self, + cooldown_secs: float = 60.0, + probe_timeout_secs: float = 30.0, + max_probe_attempts: int = 3, + sleep_fn: Callable[[float], Awaitable[None]] | None = None, + ) -> None: + self._open = True + self._close_reason: str | None = None + self._close_time: float | None = None + self._cooldown_secs = cooldown_secs + self._probe_timeout_secs = probe_timeout_secs + self._max_probe_attempts = max_probe_attempts + self._sleep: Callable[[float], Awaitable[None]] = ( + sleep_fn if sleep_fn is not None else asyncio.sleep + ) + self._recovery_task: asyncio.Task[None] | None = None + + @property + def is_open(self) -> bool: + return self._open + + @property + def close_reason(self) -> str | None: + return self._close_reason + + async def wait_if_closed(self) -> None: + """Always yield to event loop; block until open if currently closed.""" + await asyncio.sleep(0) + while not self._open: + await self._sleep(_POLL_INTERVAL) + + def signal_rate_limit( + self, + reason: str = "rate_limit", + probe_fn: Callable[[], Awaitable[bool]] | None = None, + ) -> None: + """Close the gate and schedule cooldown+probe recovery. + + Idempotent: duplicate signals while a recovery task is running are + silently dropped to prevent overlapping cooldown/probe tasks. + """ + if ( + not self._open + and self._recovery_task is not None + and not self._recovery_task.done() + ): + _log.debug( + "rate_limit_gate: already closed (recovery running), ignoring" + ) + return + self._open = False + self._close_reason = reason + self._close_time = time.monotonic() + _log.warning("rate_limit_gate: CLOSED | reason=%s", reason) + self._recovery_task = asyncio.create_task( + self._run_recovery(probe_fn), + name="rate-limit-gate-recovery", + ) + + async def _run_recovery( + self, + probe_fn: Callable[[], Awaitable[bool]] | None, + ) -> None: + try: + _log.info( + "rate_limit_gate: cooldown started (%.0fs)", self._cooldown_secs + ) + await self._sleep(self._cooldown_secs) + _log.info("rate_limit_gate: cooldown complete") + + if probe_fn is None: + self._open = True + _log.info("rate_limit_gate: REOPENED (no probe configured)") + return + + for attempt in range(1, self._max_probe_attempts + 1): + ready = False + try: + ready = await asyncio.wait_for( + probe_fn(), timeout=self._probe_timeout_secs + ) + except TimeoutError: + _log.warning( + "rate_limit_gate: probe TIMEOUT (attempt %d/%d)", + attempt, + self._max_probe_attempts, + ) + except Exception as _exc: # noqa: BLE001 + _log.warning( + "rate_limit_gate: probe EXCEPTION (attempt %d/%d): %s", + attempt, + self._max_probe_attempts, + type(_exc).__name__, + ) + + if ready: + self._open = True + _log.info( + "rate_limit_gate: REOPENED (probe passed, attempt %d/%d)", + attempt, + self._max_probe_attempts, + ) + return + + _log.warning( + "rate_limit_gate: probe FAILED (attempt %d/%d)", + attempt, + self._max_probe_attempts, + ) + if attempt < self._max_probe_attempts: + backoff = min(self._cooldown_secs, 300.0) + _log.info( + "rate_limit_gate: backing off %.0fs before next probe", + backoff, + ) + await self._sleep(backoff) + + _log.error( + "rate_limit_gate: all %d probes failed — gate remains CLOSED", + self._max_probe_attempts, + ) + except asyncio.CancelledError: + _log.debug("rate_limit_gate: recovery task cancelled") + raise + + def cancel_recovery(self) -> None: + """Cancel an in-flight recovery task and reopen the gate. + + Called when the engine that supplied the probe_fn is torn down. + Cancelling prevents the probe from running against a dead engine (which + would exhaust all attempts and leave the gate permanently closed). + Reopening ensures workers can resume once a new engine is started. + The next rate-limit signal will start a fresh recovery with the new + engine's probe_fn. + """ + if self._recovery_task is not None and not self._recovery_task.done(): + self._recovery_task.cancel() + self._open = True + _log.info("rate_limit_gate: recovery cancelled (engine teardown), gate REOPENED") # noqa: E501 diff --git a/infrastructure/persistence/repositories/player_info_queue.py b/infrastructure/persistence/repositories/player_info_queue.py index d838fbc..fa6ef46 100644 --- a/infrastructure/persistence/repositories/player_info_queue.py +++ b/infrastructure/persistence/repositories/player_info_queue.py @@ -5,7 +5,7 @@ import logging from datetime import UTC, datetime -from sqlalchemy import text +from sqlalchemy import text, update from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from core.application.retry_policy import is_terminal @@ -57,6 +57,32 @@ async def mark_failed(self, job_id: int, error: str) -> None: row.status = ScrapeStatus.PENDING await self._session.flush() + async def release_to_pending(self, job_id: int) -> bool: + """Release an IN_PROGRESS job back to PENDING without penalty. + + Used when the rate-limit gate closes between claim-at-handoff and + navigation. Does NOT increment retry_count or advance failure state. + Only transitions IN_PROGRESS → PENDING; skips any other state silently. + + Returns True if the row was updated, False if job was not IN_PROGRESS + or does not exist (safe to ignore in both cases). + """ + async with repo_error_context( + "release_to_pending", "release_to_pending failed" + ): + stmt = ( + update(ScrapeQueue) + .where( + ScrapeQueue.id == job_id, + ScrapeQueue.status == ScrapeStatus.IN_PROGRESS, + ScrapeQueue.job_type == self._job_type, + ) + .values(status=ScrapeStatus.PENDING, locked_at=None) + .returning(ScrapeQueue.id) + ) + result = await self._session.execute(stmt) + return result.fetchone() is not None + async def recover_failed_job( job_id: int, diff --git a/scripts/scrape_player_info.py b/scripts/scrape_player_info.py index 5c58b48..f4b0e3b 100644 --- a/scripts/scrape_player_info.py +++ b/scripts/scrape_player_info.py @@ -33,6 +33,7 @@ BoundedCandidateBuffer, CandidateProducer, ) +from infrastructure.browser.gate import RateLimitGate from infrastructure.display.worker_display import build_worker_table, run_display_loop from infrastructure.persistence.repositories.backend_urls import BackendUrlRepository from infrastructure.persistence.repositories.player_info import PlayerInfoRepository @@ -238,6 +239,7 @@ def __init__( step2_done: asyncio.Event | None = None, max_queue_retries: int = 5, display: XvfbDisplay | None = None, + rate_limit_gate: RateLimitGate | None = None, ) -> None: super().__init__( worker_id=worker_id, @@ -254,6 +256,7 @@ def __init__( self._step2_done = step2_done self._max_queue_retries: int = max_queue_retries self._display: XvfbDisplay | None = display + self._rate_limit_gate: RateLimitGate | None = rate_limit_gate async def on_browser_ready(self, engine: Any) -> None: from ports.browser import WarmableEngine @@ -491,6 +494,11 @@ async def run_buffered(self, buffer: BoundedCandidateBuffer) -> int: raise except Exception: restart_count += 1 + # Engine is torn down. Cancel any in-flight gate recovery that + # captured this engine's probe_fn — a dead-engine probe would + # exhaust all attempts and leave the gate permanently closed. + if self._rate_limit_gate is not None: + self._rate_limit_gate.cancel_recovery() if restart_count >= _MAX_RESTARTS: self._labels[self._worker_id] = ( f"[bold red]ERROR[/] Browser failed" @@ -507,6 +515,10 @@ async def run_buffered(self, buffer: BoundedCandidateBuffer) -> int: continue if _cooldown_required: + # Engine torn down. Cancel gate recovery so next engine session + # can start a fresh probe rather than probing a dead engine. + if self._rate_limit_gate is not None: + self._rate_limit_gate.cancel_recovery() self._labels[self._worker_id] = ( f"__cooldown__{time.monotonic() + 60}" ) @@ -525,6 +537,9 @@ async def _run_buffered_loop( self._labels[self._worker_id] = "Starting buffered crawl..." while True: + # Check gate before blocking on buffer.get() — unblocks quickly if closed. + if self._rate_limit_gate is not None: + await self._rate_limit_gate.wait_if_closed() # Guard: if buffer is already closed and empty (e.g., this is a # browser restart and the sentinel was consumed in the prior session), # exit immediately instead of blocking on get() forever. @@ -552,6 +567,23 @@ async def _run_buffered_loop( ) continue + # Gate-after-claim race: if gate closed between claim and navigation, + # release the job back to PENDING without a retry penalty. + if self._rate_limit_gate is not None and not self._rate_limit_gate.is_open: + async with get_session(self._session_factory) as _rs: + _r_repo = PlayerInfoQueueRepository( + _rs, max_queue_retries=self._max_queue_retries + ) + released = await _r_repo.release_to_pending(job.id) + await _rs.commit() + if released: + logger.info( + "rate_limit_gate: released job %d to PENDING (gate closed)", + job.id, + ) + await self._rate_limit_gate.wait_if_closed() + continue + attempt = 0 success = False browser_restart = False @@ -559,6 +591,8 @@ async def _run_buffered_loop( while attempt < 3 and not success: try: + if self._rate_limit_gate is not None: + await self._rate_limit_gate.wait_if_closed() async with self._fetch_gate: await engine.navigate(job.url) await asyncio.sleep(random.uniform(2.0, 6.0)) @@ -629,6 +663,30 @@ async def _run_buffered_loop( except (PageLoadError, RateLimitError, BrowserException) as exc: attempt += 1 + if ( + isinstance(exc, RateLimitError) + and self._rate_limit_gate is not None + ): + _engine_ref = engine # capture for closure + _fbref_url = self._fbref_base_url + + async def _probe_fn( + _e: Any = _engine_ref, + _url: str = _fbref_url, + ) -> bool: + try: + from ports.browser import WarmableEngine + if isinstance(_e, WarmableEngine): + await _e.warmup(_url) + return True + return True + except Exception: # noqa: BLE001 + return False + + self._rate_limit_gate.signal_rate_limit( + reason="rate_limit_429", + probe_fn=_probe_fn, + ) if isinstance(exc, BrowserException): self._labels[self._worker_id] = ( "[bold red]ERROR[/] Browser error — Restarting" @@ -960,6 +1018,19 @@ async def main(workers: int | None = None) -> None: ) ) if settings.scraping.player_info_dispatch_buffer_enabled: + _sc = settings.scraping + _gate = RateLimitGate( + cooldown_secs=_sc.player_info_gate_cooldown_secs, + probe_timeout_secs=_sc.player_info_gate_probe_timeout_secs, + max_probe_attempts=_sc.player_info_gate_max_probe_attempts, + ) + logger.info( + "player_info: buffered mode | workers=%d | buffer_size=%d" + " | gate_cooldown=%.0fs", + workers, + settings.scraping.player_info_dispatch_buffer_size, + settings.scraping.player_info_gate_cooldown_secs, + ) _dispatch_buffer = BoundedCandidateBuffer( maxsize=settings.scraping.player_info_dispatch_buffer_size ) @@ -1000,6 +1071,7 @@ async def _peek_pending() -> list[int]: fbref_base_url=settings.scraping.fbref_base_url, max_queue_retries=settings.scraping.max_queue_retries, display=shared_display, + rate_limit_gate=_gate, ) for i in range(workers) ] @@ -1015,6 +1087,10 @@ async def _peek_pending() -> list[int]: _producer.request_stop() await _producer_task else: + logger.info( + "player_info: direct mode | workers=%d", + workers, + ) results = await asyncio.gather( *[ PlayerInfoWorker( diff --git a/tests/integration/test_player_info_queue_release.py b/tests/integration/test_player_info_queue_release.py new file mode 100644 index 0000000..1a8a203 --- /dev/null +++ b/tests/integration/test_player_info_queue_release.py @@ -0,0 +1,119 @@ +"""Integration tests for PlayerInfoQueueRepository.release_to_pending(). + +Uses a real PostgreSQL database via testcontainers. These tests verify +that release_to_pending() transitions IN_PROGRESS → PENDING without +incrementing retry counters or advancing failure state. + +All tests use async_session from integration/conftest.py (rolled back after each test). +""" +from __future__ import annotations + +import pytest +from sqlalchemy.ext.asyncio import AsyncSession + +from infrastructure.persistence.models.scrape_queue import ScrapeQueue, ScrapeStatus +from infrastructure.persistence.repositories.player_info_queue import ( + PlayerInfoQueueRepository, +) + +pytestmark = pytest.mark.asyncio + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +async def _insert_pending( + session: AsyncSession, + url: str = "https://fbref.com/en/players/abc123/test-player", +) -> ScrapeQueue: + """Insert one PENDING player_info job; flushes but does not commit.""" + row = ScrapeQueue( + url=url, + domain="fbref.com", + job_type="player_info", + status=ScrapeStatus.PENDING, + ) + session.add(row) + await session.flush() + await session.refresh(row) + return row + + +async def _claim(session: AsyncSession, job_id: int) -> None: + repo = PlayerInfoQueueRepository(session) + row = await repo.claim_by_id(job_id) + assert row is not None + await session.flush() + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +async def test_release_to_pending_transitions_in_progress( + async_session: AsyncSession, +) -> None: + row = await _insert_pending(async_session) + await _claim(async_session, row.id) + + # Refresh to confirm IN_PROGRESS + await async_session.refresh(row) + assert row.status == ScrapeStatus.IN_PROGRESS + + repo = PlayerInfoQueueRepository(async_session) + released = await repo.release_to_pending(row.id) + await async_session.flush() + + assert released is True + await async_session.refresh(row) + assert row.status == ScrapeStatus.PENDING + assert row.locked_at is None + + +async def test_release_to_pending_does_not_increment_retry_count( + async_session: AsyncSession, +) -> None: + row = await _insert_pending( + async_session, + url="https://fbref.com/en/players/def456/retry-player", + ) + original_retry = row.retry_count + await _claim(async_session, row.id) + + repo = PlayerInfoQueueRepository(async_session) + await repo.release_to_pending(row.id) + await async_session.flush() + + await async_session.refresh(row) + assert row.retry_count == original_retry + + +async def test_release_to_pending_noop_for_pending_job( + async_session: AsyncSession, +) -> None: + """Releasing a PENDING job (not IN_PROGRESS) returns False.""" + row = await _insert_pending( + async_session, + url="https://fbref.com/en/players/ghi789/pending-player", + ) + + repo = PlayerInfoQueueRepository(async_session) + released = await repo.release_to_pending(row.id) + await async_session.flush() + + assert released is False + await async_session.refresh(row) + assert row.status == ScrapeStatus.PENDING + + +async def test_release_to_pending_noop_for_nonexistent_job( + async_session: AsyncSession, +) -> None: + repo = PlayerInfoQueueRepository(async_session) + released = await repo.release_to_pending(99999999) + await async_session.flush() + + assert released is False diff --git a/tests/unit/infrastructure/browser/test_dispatch_buffer.py b/tests/unit/infrastructure/browser/test_dispatch_buffer.py index e2d2217..55b182a 100644 --- a/tests/unit/infrastructure/browser/test_dispatch_buffer.py +++ b/tests/unit/infrastructure/browser/test_dispatch_buffer.py @@ -476,14 +476,11 @@ def test_dispatch_module_no_persistence_import(): assert "infrastructure.persistence" not in source -def test_pr6_rate_limit_gate_not_implemented(): - """Verify no rate_limit_gate attribute exists on dispatch module.""" +def test_dispatch_module_gate_is_separate_module(): + """RateLimitGate lives in gate.py, not dispatch.py — keeps dispatch focused.""" import infrastructure.browser.dispatch as dispatch - assert not hasattr(dispatch, "RateLimitGate") - assert not hasattr(dispatch, "rate_limit_gate") - + import infrastructure.browser.gate as gate_module -def test_pr6_cooldown_gate_not_implemented(): - import infrastructure.browser.dispatch as dispatch - assert not hasattr(dispatch, "CooldownGate") - assert not hasattr(dispatch, "cooldown_gate") + # gate is its own module, not merged into dispatch + assert not hasattr(dispatch, "RateLimitGate") + assert hasattr(gate_module, "RateLimitGate") diff --git a/tests/unit/infrastructure/browser/test_rate_limit_gate.py b/tests/unit/infrastructure/browser/test_rate_limit_gate.py new file mode 100644 index 0000000..c1897ed --- /dev/null +++ b/tests/unit/infrastructure/browser/test_rate_limit_gate.py @@ -0,0 +1,301 @@ +"""Unit tests for RateLimitGate.""" +from __future__ import annotations + +import asyncio +import logging +from collections.abc import Awaitable, Callable + +import pytest + +from infrastructure.browser.gate import RateLimitGate + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def instant_sleep() -> tuple[list[float], Callable[[float], Awaitable[None]]]: + """Returns (recorded, sleep_fn) where sleep_fn records delays without sleeping.""" + delays: list[float] = [] + + async def _sleep(secs: float) -> None: + delays.append(secs) + + return delays, _sleep + + +# --------------------------------------------------------------------------- +# Initial state +# --------------------------------------------------------------------------- + + +def test_gate_starts_open(): + _, sleep = instant_sleep() + gate = RateLimitGate(sleep_fn=sleep) + assert gate.is_open is True + + +def test_gate_close_reason_none_initially(): + _, sleep = instant_sleep() + gate = RateLimitGate(sleep_fn=sleep) + assert gate.close_reason is None + + +# --------------------------------------------------------------------------- +# wait_if_closed yields even when open +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_wait_if_closed_yields_when_open(): + _, sleep = instant_sleep() + gate = RateLimitGate(sleep_fn=sleep) + yielded = False + + async def check(): + nonlocal yielded + await gate.wait_if_closed() + yielded = True + + await check() + assert yielded + + +# --------------------------------------------------------------------------- +# signal_rate_limit closes gate +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_signal_rate_limit_closes_gate(): + _, sleep = instant_sleep() + gate = RateLimitGate(cooldown_secs=0.01, sleep_fn=sleep) + gate.signal_rate_limit(reason="test_close") + assert gate.is_open is False + assert gate.close_reason == "test_close" + # Cancel the recovery task + if gate._recovery_task: + gate._recovery_task.cancel() + try: + await gate._recovery_task + except (asyncio.CancelledError, Exception): + pass + + +# --------------------------------------------------------------------------- +# Cooldown + auto-reopen (no probe) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_gate_reopens_after_cooldown_no_probe(): + delays, sleep = instant_sleep() + gate = RateLimitGate(cooldown_secs=1.0, sleep_fn=sleep) + gate.signal_rate_limit(reason="rl") + assert gate.is_open is False + await gate._recovery_task # type: ignore[union-attr] + assert gate.is_open is True + assert delays[0] == 1.0 # cooldown sleep was called + + +# --------------------------------------------------------------------------- +# Cooldown + probe pass → reopen +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_gate_reopens_after_probe_pass(): + _, sleep = instant_sleep() + gate = RateLimitGate(cooldown_secs=1.0, probe_timeout_secs=5.0, sleep_fn=sleep) + + async def good_probe() -> bool: + return True + + gate.signal_rate_limit(reason="rl", probe_fn=good_probe) + await gate._recovery_task # type: ignore[union-attr] + assert gate.is_open is True + + +# --------------------------------------------------------------------------- +# Probe fail all attempts → gate remains closed +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_gate_remains_closed_after_all_probes_fail(): + _, sleep = instant_sleep() + gate = RateLimitGate( + cooldown_secs=1.0, + probe_timeout_secs=5.0, + max_probe_attempts=3, + sleep_fn=sleep, + ) + + async def bad_probe() -> bool: + return False + + gate.signal_rate_limit(reason="rl", probe_fn=bad_probe) + await gate._recovery_task # type: ignore[union-attr] + assert gate.is_open is False + + +# --------------------------------------------------------------------------- +# Probe timeout → treated as failure +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_probe_timeout_counts_as_failure(): + _, sleep = instant_sleep() + gate = RateLimitGate( + cooldown_secs=0.1, + probe_timeout_secs=0.001, + max_probe_attempts=1, + sleep_fn=sleep, + ) + + async def slow_probe() -> bool: + await asyncio.sleep(10) + return True + + gate.signal_rate_limit(reason="rl", probe_fn=slow_probe) + await gate._recovery_task # type: ignore[union-attr] + assert gate.is_open is False + + +# --------------------------------------------------------------------------- +# Idempotency: duplicate signals while recovery runs are dropped +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_duplicate_signal_while_recovery_noop(): + _, sleep = instant_sleep() + gate = RateLimitGate(cooldown_secs=1.0, sleep_fn=sleep) + gate.signal_rate_limit(reason="first") + task1 = gate._recovery_task + gate.signal_rate_limit(reason="second") + task2 = gate._recovery_task + # Both signals should share the same recovery task + assert task1 is task2 + if task1: + task1.cancel() + try: + await task1 + except (asyncio.CancelledError, Exception): + pass + + +# --------------------------------------------------------------------------- +# Re-signal after all probes exhausted starts a new recovery +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_resignal_after_all_probes_exhausted_starts_new_recovery(): + """After all probes fail, a new signal_rate_limit starts a fresh recovery.""" + call_count = 0 + + async def eventually_good() -> bool: + nonlocal call_count + call_count += 1 + return call_count > 1 # fail first call, pass second + + _, sleep = instant_sleep() + gate = RateLimitGate( + cooldown_secs=0.0, max_probe_attempts=1, sleep_fn=sleep + ) + + # First signal: probe fails → gate remains closed + gate.signal_rate_limit(reason="rl1", probe_fn=eventually_good) + await gate._recovery_task # type: ignore[union-attr] + assert gate.is_open is False + assert gate._recovery_task is not None + assert gate._recovery_task.done() + + # Second signal: probe now passes → gate reopens + gate.signal_rate_limit(reason="rl2", probe_fn=eventually_good) + await gate._recovery_task # type: ignore[union-attr] + assert gate.is_open is True + + +# --------------------------------------------------------------------------- +# cancel_recovery() reopens gate and handles no-op when no task running +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_cancel_recovery_reopens_gate(): + _, sleep = instant_sleep() + gate = RateLimitGate(cooldown_secs=100.0, sleep_fn=sleep) + gate.signal_rate_limit(reason="rl") + assert gate.is_open is False + gate.cancel_recovery() + assert gate.is_open is True + # Recovery task was cancelled — await it to clean up + if gate._recovery_task is not None: + try: + await gate._recovery_task + except (asyncio.CancelledError, Exception): + pass + + +def test_cancel_recovery_noop_when_no_task(): + _, sleep = instant_sleep() + gate = RateLimitGate(sleep_fn=sleep) + gate.cancel_recovery() # must not raise + assert gate.is_open is True + + +# --------------------------------------------------------------------------- +# CancelledError propagates cleanly +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_recovery_task_cancel(): + _, sleep = instant_sleep() + gate = RateLimitGate(cooldown_secs=100.0, sleep_fn=sleep) + gate.signal_rate_limit(reason="rl") + assert gate._recovery_task is not None + gate._recovery_task.cancel() + with pytest.raises(asyncio.CancelledError): + await gate._recovery_task + + +# --------------------------------------------------------------------------- +# Non-rate-limit exceptions do not close the gate +# --------------------------------------------------------------------------- + + +def test_no_auto_close_on_creation(): + _, sleep = instant_sleep() + gate = RateLimitGate(sleep_fn=sleep) + assert gate.is_open is True # fresh gate is always open + + +# --------------------------------------------------------------------------- +# Security: no secrets in log records +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_gate_logs_no_sensitive_fields(caplog: pytest.LogCaptureFixture): + _, sleep = instant_sleep() + gate = RateLimitGate(cooldown_secs=0.01, sleep_fn=sleep) + with caplog.at_level(logging.DEBUG, logger="infrastructure.browser.gate"): + gate.signal_rate_limit(reason="rate_limit_429") + if gate._recovery_task: + gate._recovery_task.cancel() + try: + await gate._recovery_task + except (asyncio.CancelledError, Exception): + pass + for record in caplog.records: + msg = record.getMessage().lower() + assert "cookie" not in msg + assert "token" not in msg + assert "password" not in msg + assert "ws://" not in msg + assert "wss://" not in msg + assert "cdp" not in msg diff --git a/tests/unit/infrastructure/player_info/__init__.py b/tests/unit/infrastructure/player_info/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/infrastructure/player_info/test_mode_selection.py b/tests/unit/infrastructure/player_info/test_mode_selection.py new file mode 100644 index 0000000..ba5a93a --- /dev/null +++ b/tests/unit/infrastructure/player_info/test_mode_selection.py @@ -0,0 +1,104 @@ +"""Unit tests for player_info mode selection (direct vs buffered).""" +from __future__ import annotations + +from config.settings import ScrapingSettings + +# --------------------------------------------------------------------------- +# Feature flag defaults +# --------------------------------------------------------------------------- + + +def test_default_buffer_disabled(): + """Buffered mode must be off by default.""" + s = ScrapingSettings() + assert s.player_info_dispatch_buffer_enabled is False + + +def test_default_warm_pool_disabled(): + s = ScrapingSettings() + assert s.player_info_warm_pool_enabled is False + + +def test_default_gate_cooldown(): + s = ScrapingSettings() + assert s.player_info_gate_cooldown_secs >= 5.0 + + +def test_default_gate_probe_timeout(): + s = ScrapingSettings() + assert s.player_info_gate_probe_timeout_secs >= 5.0 + + +def test_default_gate_max_probe_attempts(): + s = ScrapingSettings() + assert s.player_info_gate_max_probe_attempts >= 1 + + +def test_gate_settings_respect_minimum(): + """Settings must reject values below their floor.""" + import pytest + from pydantic import ValidationError + with pytest.raises(ValidationError): + ScrapingSettings(player_info_gate_cooldown_secs=1.0) # below ge=5.0 + + +# --------------------------------------------------------------------------- +# Mutual exclusion (logic-level assertions via settings inspection) +# --------------------------------------------------------------------------- + + +def test_direct_and_buffered_modes_are_mutually_exclusive_via_flag(): + """The single boolean flag determines mode — only one can be active.""" + s_direct = ScrapingSettings(player_info_dispatch_buffer_enabled=False) + s_buffered = ScrapingSettings(player_info_dispatch_buffer_enabled=True) + # Verify the flag round-trips correctly — mode is selected by this single bool. + assert s_direct.player_info_dispatch_buffer_enabled is False + assert s_buffered.player_info_dispatch_buffer_enabled is True + # A single process uses exactly one config: flag cannot be both True and False. + assert s_direct.player_info_dispatch_buffer_enabled != ( + s_buffered.player_info_dispatch_buffer_enabled + ) + + +# --------------------------------------------------------------------------- +# Gate integration: RateLimitGate is only used in buffered mode +# --------------------------------------------------------------------------- + + +def test_rate_limit_gate_module_imports_cleanly(): + from infrastructure.browser.gate import RateLimitGate + gate = RateLimitGate() + assert gate.is_open is True + + +def test_gate_no_persistence_import(): + """gate.py must not import from infrastructure.persistence.""" + import importlib + import inspect + import sys + + mod_name = "infrastructure.browser.gate" + if mod_name in sys.modules: + mod = sys.modules[mod_name] + else: + mod = importlib.import_module(mod_name) + + source = inspect.getsource(mod) + assert "infrastructure.persistence" not in source + + +def test_gate_no_config_import(): + """gate.py must not import from config — settings are injected by orchestrator.""" + import importlib + import inspect + import sys + + mod_name = "infrastructure.browser.gate" + if mod_name in sys.modules: + mod = sys.modules[mod_name] + else: + mod = importlib.import_module(mod_name) + + source = inspect.getsource(mod) + assert "config.settings" not in source + assert "from config" not in source From 055c285aa4ab8ed03f9c5b1222a82a9196797b9d Mon Sep 17 00:00:00 2001 From: ChechiDev Date: Sun, 2 Aug 2026 15:58:01 +0200 Subject: [PATCH 2/2] fix(player-info): harden buffered gate integration - cancel_recovery() no longer blindly reopens gate; gate stays CLOSED until new engine proves readiness via mark_engine_ready() - Add mark_engine_ready(): called after on_browser_ready()/warmup() succeeds; also cancels any residual stale recovery task - Add shutdown(): cancels recovery task on clean process exit without changing gate state; called in main() finally block - Add _MAX_PROBE_BACKOFF_SECS = 300.0 module-level constant - Add cancel_recovery() to inner except block in run_buffered() so stale probes are cancelled when on_browser_ready/loop throws unexpectedly - Fix probe_fn for non-WarmableEngine: pass probe_fn=None instead of returning True blindly (gate reopens after cooldown, not immediately) - Pass wait_fn=_gate.wait_if_closed to CandidateProducer so producer explicitly pauses while gate is closed (not just via backpressure) - Add execution_options(synchronize_session=False) to release_to_pending UPDATE to prevent identity-map stale-read trap - Add test: wait_if_closed blocks while gate is closed and unblocks on mark_engine_ready - Add tests: producer respects wait_fn; cancel_recovery leaves gate closed; mark_engine_ready opens gate; shutdown is idempotent - Add integration tests: release_to_pending is noop for DONE and FAILED rows --- infrastructure/browser/dispatch.py | 4 + infrastructure/browser/gate.py | 35 ++++-- .../repositories/player_info_queue.py | 1 + scripts/scrape_player_info.py | 34 ++++-- .../test_player_info_queue_release.py | 44 +++++++ .../browser/test_dispatch_buffer.py | 39 +++++++ .../browser/test_rate_limit_gate.py | 107 +++++++++++++++++- 7 files changed, 241 insertions(+), 23 deletions(-) diff --git a/infrastructure/browser/dispatch.py b/infrastructure/browser/dispatch.py index fa58f14..8242881 100644 --- a/infrastructure/browser/dispatch.py +++ b/infrastructure/browser/dispatch.py @@ -153,6 +153,7 @@ def __init__( n_workers: int, poll_interval: float = 5.0, step2_done: asyncio.Event | None = None, + wait_fn: Callable[[], Awaitable[None]] | None = None, ) -> None: if CandidateProducer._active: raise RuntimeError( @@ -164,6 +165,7 @@ def __init__( self._n_workers = n_workers self._poll_interval = poll_interval self._step2_done = step2_done + self._wait_fn = wait_fn self._stop = asyncio.Event() def request_stop(self) -> None: @@ -179,6 +181,8 @@ async def run(self) -> None: """Run the producer loop until the queue is exhausted or stop is requested.""" try: while not self._stop.is_set(): + if self._wait_fn is not None: + await self._wait_fn() ids = await self._peek_fn() # Transaction inside peek_fn is closed before put() is called. if ids: diff --git a/infrastructure/browser/gate.py b/infrastructure/browser/gate.py index b9314b2..3d6ccdf 100644 --- a/infrastructure/browser/gate.py +++ b/infrastructure/browser/gate.py @@ -16,6 +16,7 @@ _log = logging.getLogger(__name__) _POLL_INTERVAL: float = 1.0 +_MAX_PROBE_BACKOFF_SECS: float = 300.0 class RateLimitGate: @@ -77,6 +78,9 @@ def signal_rate_limit( ) -> None: """Close the gate and schedule cooldown+probe recovery. + Must be called from within a running asyncio event loop because it + calls asyncio.create_task() to schedule the recovery coroutine. + Idempotent: duplicate signals while a recovery task is running are silently dropped to prevent overlapping cooldown/probe tasks. """ @@ -149,7 +153,7 @@ async def _run_recovery( self._max_probe_attempts, ) if attempt < self._max_probe_attempts: - backoff = min(self._cooldown_secs, 300.0) + backoff = min(self._cooldown_secs, _MAX_PROBE_BACKOFF_SECS) _log.info( "rate_limit_gate: backing off %.0fs before next probe", backoff, @@ -165,16 +169,31 @@ async def _run_recovery( raise def cancel_recovery(self) -> None: - """Cancel an in-flight recovery task and reopen the gate. + """Cancel an in-flight recovery task without reopening the gate. Called when the engine that supplied the probe_fn is torn down. - Cancelling prevents the probe from running against a dead engine (which - would exhaust all attempts and leave the gate permanently closed). - Reopening ensures workers can resume once a new engine is started. - The next rate-limit signal will start a fresh recovery with the new - engine's probe_fn. + Cancelling prevents exhausting probe attempts against a dead engine. + The gate remains CLOSED until the next engine proves readiness via + mark_engine_ready(). + """ + if self._recovery_task is not None and not self._recovery_task.done(): + self._recovery_task.cancel() + _log.debug("rate_limit_gate: recovery task cancelled (engine teardown), gate remains CLOSED") # noqa: E501 + + def mark_engine_ready(self) -> None: + """Open the gate after a new engine session has proven readiness. + + Called after a successful on_browser_ready()/warmup() so workers can + resume navigation. Also cancels any residual recovery task from a + previous engine session. """ if self._recovery_task is not None and not self._recovery_task.done(): self._recovery_task.cancel() self._open = True - _log.info("rate_limit_gate: recovery cancelled (engine teardown), gate REOPENED") # noqa: E501 + _log.info("rate_limit_gate: REOPENED (new engine ready via warmup)") + + def shutdown(self) -> None: + """Cancel any in-flight recovery task on process exit (no gate state change).""" + if self._recovery_task is not None and not self._recovery_task.done(): + self._recovery_task.cancel() + _log.debug("rate_limit_gate: shutdown, recovery task cancelled if running") diff --git a/infrastructure/persistence/repositories/player_info_queue.py b/infrastructure/persistence/repositories/player_info_queue.py index fa6ef46..a7368ec 100644 --- a/infrastructure/persistence/repositories/player_info_queue.py +++ b/infrastructure/persistence/repositories/player_info_queue.py @@ -79,6 +79,7 @@ async def release_to_pending(self, job_id: int) -> bool: ) .values(status=ScrapeStatus.PENDING, locked_at=None) .returning(ScrapeQueue.id) + .execution_options(synchronize_session=False) ) result = await self._session.execute(stmt) return result.fetchone() is not None diff --git a/scripts/scrape_player_info.py b/scripts/scrape_player_info.py index f4b0e3b..1fb2782 100644 --- a/scripts/scrape_player_info.py +++ b/scripts/scrape_player_info.py @@ -12,6 +12,7 @@ import logging import random import time +from collections.abc import Awaitable, Callable from typing import TYPE_CHECKING, Any if TYPE_CHECKING: @@ -476,6 +477,9 @@ async def run_buffered(self, buffer: BoundedCandidateBuffer) -> int: try: restart_count = 0 await self.on_browser_ready(engine) + # Gate was closed by rate-limit; warmup just proved readiness. + if self._rate_limit_gate is not None: + self._rate_limit_gate.mark_engine_ready() loop_result = await self._run_buffered_loop(engine, buffer) except CooldownRequired: _cooldown_required = True @@ -483,6 +487,8 @@ async def run_buffered(self, buffer: BoundedCandidateBuffer) -> int: except asyncio.CancelledError: raise except Exception as exc: + if self._rate_limit_gate is not None: + self._rate_limit_gate.cancel_recovery() _w_log.error(exc, exc_info=True) self._labels[self._worker_id] = "unexpected error — restarting" await asyncio.sleep(5) @@ -670,22 +676,26 @@ async def _run_buffered_loop( _engine_ref = engine # capture for closure _fbref_url = self._fbref_base_url - async def _probe_fn( - _e: Any = _engine_ref, - _url: str = _fbref_url, - ) -> bool: - try: - from ports.browser import WarmableEngine - if isinstance(_e, WarmableEngine): + from ports.browser import WarmableEngine as _WarmableEngine + _probe: Callable[[], Awaitable[bool]] | None = None + if isinstance(_engine_ref, _WarmableEngine): + _e_ref = _engine_ref + _u_ref = _fbref_url + + async def _probe_fn( # noqa: E306 + _e: Any = _e_ref, + _url: str = _u_ref, + ) -> bool: + try: await _e.warmup(_url) return True - return True - except Exception: # noqa: BLE001 - return False + except Exception: # noqa: BLE001 + return False + _probe = _probe_fn self._rate_limit_gate.signal_rate_limit( reason="rate_limit_429", - probe_fn=_probe_fn, + probe_fn=_probe, ) if isinstance(exc, BrowserException): self._labels[self._worker_id] = ( @@ -1053,6 +1063,7 @@ async def _peek_pending() -> list[int]: poll_interval=( settings.scraping.player_info_dispatch_buffer_poll_interval ), + wait_fn=_gate.wait_if_closed, ) _producer_task = asyncio.create_task( _producer.run(), name="dispatch-producer" @@ -1086,6 +1097,7 @@ async def _peek_pending() -> list[int]: finally: _producer.request_stop() await _producer_task + _gate.shutdown() else: logger.info( "player_info: direct mode | workers=%d", diff --git a/tests/integration/test_player_info_queue_release.py b/tests/integration/test_player_info_queue_release.py index 1a8a203..86415d2 100644 --- a/tests/integration/test_player_info_queue_release.py +++ b/tests/integration/test_player_info_queue_release.py @@ -117,3 +117,47 @@ async def test_release_to_pending_noop_for_nonexistent_job( await async_session.flush() assert released is False + + +async def test_release_to_pending_noop_for_done_job( + async_session: AsyncSession, +) -> None: + """release_to_pending() must not touch a DONE job.""" + job = ScrapeQueue( + url="https://fbref.com/en/players/done_player", + domain="fbref.com", + job_type="player_info", + status=ScrapeStatus.DONE, + ) + async_session.add(job) + await async_session.flush() + + repo = PlayerInfoQueueRepository(async_session) + released = await repo.release_to_pending(job.id) + await async_session.flush() + + assert released is False + await async_session.refresh(job) + assert job.status == ScrapeStatus.DONE + + +async def test_release_to_pending_noop_for_failed_job( + async_session: AsyncSession, +) -> None: + """release_to_pending() must not touch a FAILED job.""" + job = ScrapeQueue( + url="https://fbref.com/en/players/failed_player", + domain="fbref.com", + job_type="player_info", + status=ScrapeStatus.FAILED, + ) + async_session.add(job) + await async_session.flush() + + repo = PlayerInfoQueueRepository(async_session) + released = await repo.release_to_pending(job.id) + await async_session.flush() + + assert released is False + await async_session.refresh(job) + assert job.status == ScrapeStatus.FAILED diff --git a/tests/unit/infrastructure/browser/test_dispatch_buffer.py b/tests/unit/infrastructure/browser/test_dispatch_buffer.py index 55b182a..b126654 100644 --- a/tests/unit/infrastructure/browser/test_dispatch_buffer.py +++ b/tests/unit/infrastructure/browser/test_dispatch_buffer.py @@ -484,3 +484,42 @@ def test_dispatch_module_gate_is_separate_module(): # gate is its own module, not merged into dispatch assert not hasattr(dispatch, "RateLimitGate") assert hasattr(gate_module, "RateLimitGate") + + +# --------------------------------------------------------------------------- +# Producer gate awareness via wait_fn +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_producer_pauses_while_gate_closed(): + """Producer wait_fn is called at the start of each loop iteration. + + The producer runs until the queue is empty (no step2_done), so it executes + at least one full iteration: wait → peek → (empty → stop). This verifies + that wait_fn is always called before peek_fn. + """ + calls: list[str] = [] + + async def mock_peek() -> list[int]: + calls.append("peek") + return [] # empty — producer exits naturally (step2_done=None, buffer empty) + + async def mock_wait() -> None: + calls.append("wait") + + buf = BoundedCandidateBuffer(maxsize=10) + producer = CandidateProducer( + peek_fn=mock_peek, + buffer=buf, + n_workers=1, + poll_interval=0.01, + wait_fn=mock_wait, + ) + # Do NOT call request_stop() before run() — that would skip the loop body. + # The producer stops naturally when peek returns [] and the buffer is empty. + await asyncio.wait_for(producer.run(), timeout=2.0) + # wait was called before peek in every iteration + assert len(calls) >= 2 + assert calls[0] == "wait" + assert calls[1] == "peek" diff --git a/tests/unit/infrastructure/browser/test_rate_limit_gate.py b/tests/unit/infrastructure/browser/test_rate_limit_gate.py index c1897ed..b180f26 100644 --- a/tests/unit/infrastructure/browser/test_rate_limit_gate.py +++ b/tests/unit/infrastructure/browser/test_rate_limit_gate.py @@ -220,19 +220,20 @@ async def eventually_good() -> bool: # --------------------------------------------------------------------------- -# cancel_recovery() reopens gate and handles no-op when no task running +# cancel_recovery() leaves gate closed and handles no-op when no task running # --------------------------------------------------------------------------- @pytest.mark.asyncio -async def test_cancel_recovery_reopens_gate(): +async def test_cancel_recovery_leaves_gate_closed(): + """cancel_recovery() cancels the stale task but does NOT reopen the gate.""" _, sleep = instant_sleep() gate = RateLimitGate(cooldown_secs=100.0, sleep_fn=sleep) gate.signal_rate_limit(reason="rl") assert gate.is_open is False gate.cancel_recovery() - assert gate.is_open is True - # Recovery task was cancelled — await it to clean up + assert gate.is_open is False # gate stays closed + # Clean up cancelled task if gate._recovery_task is not None: try: await gate._recovery_task @@ -247,6 +248,57 @@ def test_cancel_recovery_noop_when_no_task(): assert gate.is_open is True +# --------------------------------------------------------------------------- +# mark_engine_ready() opens gate and cancels stale recovery +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_mark_engine_ready_opens_gate(): + """mark_engine_ready() opens a closed gate and cancels stale recovery.""" + _, sleep = instant_sleep() + gate = RateLimitGate(cooldown_secs=100.0, sleep_fn=sleep) + gate.signal_rate_limit(reason="rl") + assert gate.is_open is False + gate.mark_engine_ready() + assert gate.is_open is True + # Recovery task was cancelled + if gate._recovery_task is not None: + try: + await gate._recovery_task + except (asyncio.CancelledError, Exception): + pass + + +def test_mark_engine_ready_noop_when_already_open(): + """mark_engine_ready() on an already-open gate does not raise.""" + _, sleep = instant_sleep() + gate = RateLimitGate(sleep_fn=sleep) + gate.mark_engine_ready() # gate was already open, no recovery task + assert gate.is_open is True + + +# --------------------------------------------------------------------------- +# shutdown() cancels recovery without changing gate state +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_shutdown_cancels_recovery_task(): + """shutdown() cancels an in-flight recovery without changing gate state.""" + _, sleep = instant_sleep() + gate = RateLimitGate(cooldown_secs=100.0, sleep_fn=sleep) + gate.signal_rate_limit(reason="rl") + assert gate.is_open is False + gate.shutdown() + assert gate.is_open is False # gate state unchanged + if gate._recovery_task is not None: + try: + await gate._recovery_task + except (asyncio.CancelledError, Exception): + pass + + # --------------------------------------------------------------------------- # CancelledError propagates cleanly # --------------------------------------------------------------------------- @@ -263,6 +315,53 @@ async def test_recovery_task_cancel(): await gate._recovery_task +# --------------------------------------------------------------------------- +# wait_if_closed blocks while gate is closed and unblocks on reopen +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_wait_if_closed_blocks_then_unblocks(): + """wait_if_closed() blocks while gate is closed and unblocks when it reopens.""" + # A sleep that yields to the event loop so tasks interleave naturally. + async def yielding_sleep(_: float) -> None: + await asyncio.sleep(0) + + gate = RateLimitGate(cooldown_secs=100.0, sleep_fn=yielding_sleep) + gate.signal_rate_limit(reason="rl") + assert gate.is_open is False + + # Cancel the recovery task immediately so it cannot auto-reopen the gate. + assert gate._recovery_task is not None + gate._recovery_task.cancel() + try: + await gate._recovery_task + except (asyncio.CancelledError, Exception): + pass + + unblocked = False + + async def waiter() -> None: + nonlocal unblocked + await gate.wait_if_closed() + unblocked = True + + # Start the waiter concurrently. + task = asyncio.create_task(waiter()) + # Allow the event loop to run — waiter should be blocked in wait loop. + await asyncio.sleep(0) + await asyncio.sleep(0) + assert not unblocked # still blocked + + # Reopen the gate manually. + gate.mark_engine_ready() + await asyncio.sleep(0) + await asyncio.sleep(0) + assert unblocked # now unblocked + + await task + + # --------------------------------------------------------------------------- # Non-rate-limit exceptions do not close the gate # ---------------------------------------------------------------------------