diff --git a/CHANGELOG.md b/CHANGELOG.md index b06d4b60..1e00516b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,52 @@ accepts both, but the store flags the old spelling as deprecated ## Unreleased +### Shared fetch service (stage 1) + +Core's own HTTP fetch paths now go through one service, so the plugins that +use them get pooling, merging, host budgets and per-plugin request counts +without a code change. Return values, exceptions, cache keys, TTLs and retry +policies are unchanged. + +- **What goes through it.** `APIHelper.get`/`post`, `fetch_espn_scoreboard` + and its date chunks (`src/common/espn_dates.py` -- every scoreboard's live, + recent and upcoming fetch, and `SportsFetchMixin._fetch_season_directly`), + `BackgroundDataService` and `BaseOddsManager.get_odds`. Plugins' own + `requests` calls are not covered yet. +- **Shared connection pools.** Core sessions with the same retry policy mount + one shared adapter, so the odds managers (one per scoreboard league + manager), the background service and the APIHelpers reuse one connection + pool per host. Headers, cookies and auth stay per session. +- **Merged requests.** Identical GETs in flight at once (same URL and query, + effective headers, timeout and retry policy) go out once; the others get a + copy of that response or the same exception. `BackgroundDataService`'s own + request opts out (`share_in_flight=False`): it cancels and replaces fetches, + and already merges by cache key. +- **Host budgets.** Per-host token buckets, `fetch_service.rate_limits` in + `config.json` (new optional section in the template). ESPN hosts default to + 20 requests/s with a burst of 200, far above normal traffic; no request waits + longer than `max_wait_seconds` (2 s). Other hosts are unthrottled. +- **Conditional GET.** A response with `ETag` or `Last-Modified` is kept in a + small bounded store (64 entries, 4 MB, 1 MB each) and revalidated; a `304` + is returned to the caller as the original `200`. ESPN sends neither + validator today, so on ESPN this is dormant. +- **Counters.** Requests, merged, bytes, 304s, errors, HTTP errors, adapter + retries, throttled requests and seconds waited, per plugin and per host. + Which plugin made a request comes from a context variable the plugin + executor and plugin loader set (carried across the background service's and + `espn_dates`' worker threads), or else from the plugin directory on the + stack, so a plugin's own threads count too. The display publishes them to + the shared cache at most once a minute on change; read them at + `GET /api/v3/plugins/fetch-stats`. +- `fetch_service` is a core config section (`src/core_config_keys.py`). + +### New modules + +- `src/common/fetch_service.py` -- the fetch service above. Core-internal in + this release: plugins reach it through `APIHelper` and `espn_dates`, and + should not import it directly until a plugin-facing API ships (stage 3), so + it sets no `ledmatrix_min_version` floor. + ### Tooling - Golden trace tests for the display loop. `test/test_run_loop_golden.py` diff --git a/config/config.template.json b/config/config.template.json index 98b0f30e..f3541cf1 100644 --- a/config/config.template.json +++ b/config/config.template.json @@ -174,6 +174,16 @@ "plugin_system": { "plugins_directory": "plugin-repos" }, + "fetch_service": { + "enabled": true, + "max_wait_seconds": 2, + "rate_limits": { + "*.espn.com": { + "per_second": 20, + "burst": 200 + } + } + }, "web-ui-info": { "enabled": true, "display_duration": 10 diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 8f28883c..0a652888 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -48,6 +48,7 @@ each other. They share three things: | Plugin errors | cache `plugin_error_snapshot` | display: `ErrorSnapshotPublisher` ([`src/error_aggregator.py`](../src/error_aggregator.py)) | web: `read_error_report()` for `/api/v3/errors/*` | | Error clear | cache `plugin_error_clear_request` | web | display | | Font usage | cache `font_usage_snapshot` | display: `FontUsagePublisher` ([`src/font_usage.py`](../src/font_usage.py)) | web: Fonts tab | +| Fetch statistics (requests per plugin and host) | cache `fetch_stats_snapshot` | display: `FetchStatsPublisher` ([`src/common/fetch_service.py`](../src/common/fetch_service.py)), at most once a minute on change | web: `read_fetch_stats()` for `/api/v3/plugins/fetch-stats` | | Plugin health | cache `plugin_health:` | display (web writes on reset) | web: `/api/v3/plugins/health` | | Plugin runtime (loaded, state, last error, version) | cache `plugin_runtime_snapshot` | display: `PluginRuntimePublisher` ([`src/plugin_system/plugin_runtime.py`](../src/plugin_system/plugin_runtime.py)) | web: `read_plugin_runtime()` for `/api/v3/plugins/installed`, `/plugins/state`, reconciliation | | Preview frame | `/tmp/led_matrix_preview.png` | display: `DisplayManager`, gated by [`snapshot_policy`](../src/common/snapshot_policy.py) | web: display SSE stream, `/api/v3/health` (file age) | diff --git a/docs/PLUGIN_API_REFERENCE.md b/docs/PLUGIN_API_REFERENCE.md index f9f517b3..b3331a7a 100644 --- a/docs/PLUGIN_API_REFERENCE.md +++ b/docs/PLUGIN_API_REFERENCE.md @@ -14,6 +14,7 @@ Complete API reference for plugin developers. This document describes all method - [Display Manager](#display-manager) - [Cache Manager](#cache-manager) - [Plugin Manager](#plugin-manager) +- [Fetching data](#fetching-data) - [Deprecated APIs](#deprecated-apis) --- @@ -1030,6 +1031,63 @@ if weather is not None and weather.enabled: --- +## Fetching data + +Use the core helpers for HTTP rather than a `requests.Session` of your own: +`APIHelper` (`from src.common import APIHelper`) for JSON APIs, and +`fetch_espn_scoreboard()` (`src.common.espn_dates`) or +`BackgroundDataService` for ESPN scoreboards. Since the release after 3.7.0 +these go through the core **fetch service** (`src/common/fetch_service.py`), +so a plugin that uses them gets the following with no code change. Return +values, exceptions and retries are what they were. + +- **Shared connections.** Core sessions with the same retry policy share one + connection pool per host, instead of one pool per helper. +- **Merged requests.** Identical GETs in flight at the same time (same URL + and query, headers, timeout and retry policy) go to the network once, and + every caller gets its own copy of the response, or the same exception. +- **Host budgets.** A host can have a token-bucket budget. A request past it + waits for a token, but never longer than `max_wait_seconds` (2 s by + default). Only ESPN hosts have one by default (20 requests a second, burst + 200), which normal use never reaches. +- **Conditional GET.** When a server sends `ETag` or `Last-Modified`, the + next identical request revalidates, and a `304 Not Modified` comes back to + your code as the original `200` with its body. ESPN currently sends + neither, so this does nothing there. +- **Counters.** Requests, merged requests, bytes, 304s, errors and time spent + waiting are counted per plugin and per host, and published for the web UI + at `GET /api/v3/plugins/fetch-stats` (see + [REST_API_REFERENCE.md](REST_API_REFERENCE.md#get-fetch-statistics)). A + request is counted against your plugin when it runs inside your + `update()`/`display()`, your constructor or `on_enable()`, or anywhere in + code under your plugin's directory, including threads you start. + +What is not covered yet: requests a plugin makes with its own `requests.get()` +or `Session.get()` calls. They work as before but are invisible to the +budgets and counters. + +The settings live in `config.json` under `fetch_service`, read when the +display starts and on a config reload: + +```json +"fetch_service": { + "enabled": true, + "max_wait_seconds": 2, + "rate_limits": { + "*.espn.com": {"per_second": 20, "burst": 200}, + "api.example.com": {"per_second": 1, "burst": 5} + } +} +``` + +`rate_limits` keys are a host or a `*.domain` pattern (which also matches +the bare domain); `"per_second": 0` removes a budget. `"enabled": false` +turns the whole service into a plain `session.get()`. Two further switches, +`"single_flight": false` and `"conditional_get": false`, turn off merging and +revalidation. + +--- + ## Best Practices ### Caching diff --git a/docs/REST_API_REFERENCE.md b/docs/REST_API_REFERENCE.md index db1b0970..37479f1f 100644 --- a/docs/REST_API_REFERENCE.md +++ b/docs/REST_API_REFERENCE.md @@ -980,6 +980,63 @@ Metrics for one plugin; `data` has the same fields as one entry above. Reset metrics for a plugin. +### Get Fetch Statistics + +**GET** `/api/v3/plugins/fetch-stats` + +Network requests made through the core fetch service +(`src/common/fetch_service.py`), per plugin and per host, cumulative since +the display started. Read-only. The display publishes the counters at most +once a minute when they change (every 10 minutes otherwise), so they can be +up to a minute old. Requests a plugin makes with its own `requests` calls, +outside `APIHelper`, `espn_dates`, `BackgroundDataService` and +`BaseOddsManager`, are not counted yet. + +`data.status` is `live`, `stale` (no publish for longer than +`stale_after`), `stopped` (the display exited; the last counters are kept) +or `unknown` (nothing published; `data.data` is `null`). + +**Response**: +```json +{ + "status": "success", + "data": { + "status": "live", + "age_seconds": 12.4, + "data": { + "schema": 1, + "running": true, + "published_at": 1790000000.0, + "stale_after": 720.0, + "since": 1789990000.0, + "totals": {"requests": 412, "merged": 3, "not_modified": 0, + "errors": 1, "http_errors": 2, "retries": 0, + "throttled": 0, "overruns": 0, "bytes": 18234011, + "wait_seconds": 0.0}, + "plugins": { + "football-scoreboard": {"requests": 240, "merged": 2, "bytes": 9120330, + "hosts": {"site.api.espn.com": 180, + "sports.core.api.espn.com": 62}, + "...": "the other counters, as in totals"} + }, + "hosts": { + "site.api.espn.com": {"requests": 301, "...": "as in totals"} + }, + "validators": {"entries": 0, "bytes": 0}, + "config": {"enabled": true, "single_flight": true, + "conditional_get": true, "max_wait_seconds": 2.0, + "rate_limits": {"*.espn.com": {"per_second": 20.0, "burst": 200.0}}} + } + } +} +``` + +`requests` counts round trips sent (retries inside the HTTP adapter are in +`retries`), `merged` requests answered by an identical one already in +flight, `not_modified` 304s served from the stored body, `errors` transport +failures and `http_errors` responses with status 400 or above. `bytes` is the +decoded body size. `core` is everything no plugin made. + ### Get/Set Plugin Limits **GET** `/api/v3/plugins/limits/` diff --git a/mypy-clean.txt b/mypy-clean.txt index 80c92425..d612e716 100644 --- a/mypy-clean.txt +++ b/mypy-clean.txt @@ -22,6 +22,7 @@ src/common/api_helper.py src/common/bdf_font.py src/common/espn_dates.py src/common/favorite_team_check.py +src/common/fetch_service.py src/common/font_layout.py src/common/frame_timing.py src/common/json_body.py diff --git a/src/background_data_service.py b/src/background_data_service.py index bca6a8b5..be860bb9 100644 --- a/src/background_data_service.py +++ b/src/background_data_service.py @@ -27,6 +27,13 @@ from concurrent.futures import ThreadPoolExecutor import pytz from src.cache_manager import CacheManager from src.common.json_body import response_json +from src.common.fetch_service import ( + current_plugin_id, + fetch_get, + get_fetch_service, + plugin_scope, + share_connection_pool, +) from src.common.espn_dates import ( RANGE_RETRY_SECONDS, _note_range_rejected, @@ -78,6 +85,9 @@ class FetchRequest: commit_claimed: bool = False result: Optional[Any] = None error: Optional[str] = None + # The plugin that submitted the request, so the fetch service counts the + # worker's requests against it (fetch_service, caller identity). + owner: Optional[str] = None @dataclass class FetchResult: @@ -119,6 +129,12 @@ class _ConnectionRetryingSession: def __init__(self, session): self._session = session + @property + def fetch_identity_session(self): + """The wrapped Session, whose headers and adapter the fetch service + reads to key this request (src/common/fetch_service.py).""" + return self._session + def get(self, *args, **kwargs): for attempt in range(self.ATTEMPTS): try: @@ -196,9 +212,12 @@ class BackgroundDataService: # connection errors three times, a dead network cost up to 16 # connection attempts per request and held one of the few worker # threads for all of them. + # + # The adapter is the fetch service's shared no-retry one: the same + # max_retries=0, with the connection pool shared with the other core + # sessions that do not retry (the odds managers). self.session = requests.Session() - self.session.mount('http://', requests.adapters.HTTPAdapter(max_retries=0)) - self.session.mount('https://', requests.adapters.HTTPAdapter(max_retries=0)) + share_connection_pool(self.session, max_retries=0) # Default headers: core's shared set (real User-Agent, no hand-set # Accept-Encoding) -- see src/common/api_helper.py. @@ -299,6 +318,10 @@ class BackgroundDataService: if url.split('?', 1)[0].rstrip('/').endswith('/scoreboard'): params = clamp_espn_limit(params) + # Who asked, resolved on the submitting thread: the worker thread + # runs no plugin code, so it could not tell (fetch_service). + owner = current_plugin_id() + # Create fetch request request = FetchRequest( id=request_id, @@ -311,7 +334,8 @@ class BackgroundDataService: timeout=timeout or self.request_timeout, max_retries=max_retries, priority=priority, - callback=callback + callback=callback, + owner=owner, ) with self._lock: @@ -330,6 +354,7 @@ class BackgroundDataService: self.stats['deduplicated_requests'] = ( self.stats.get('deduplicated_requests', 0) + 1 ) + get_fetch_service().note_merged(url, owner) logger.info( "Joined in-flight fetch %s for %s (cache_key=%s) instead of " "starting a duplicate", existing_id, sport, cache_key @@ -357,6 +382,11 @@ class BackgroundDataService: Returns: Fetch result with data or error information """ + with plugin_scope(request.owner): + return self._fetch_data_worker_scoped(request) + + def _fetch_data_worker_scoped(self, request: FetchRequest) -> FetchResult: + """_fetch_data_worker's body, run with the submitter as the caller.""" start_time = time.time() result = FetchResult(request_id=request.id, success=False, retry_count=request.retry_count) @@ -621,8 +651,14 @@ class BackgroundDataService: for attempt in range(request.max_retries + 1): try: - response = self.session.get( + # Not shared with an identical request in flight: this + # service cancels and replaces fetches, and a replacement + # must not join the one it replaced. Its own cache_key + # dedup already merges what should be merged. + response = fetch_get( + self.session, request.url, + share_in_flight=False, params=request.params, headers=request.headers, timeout=request.timeout diff --git a/src/base_odds_manager.py b/src/base_odds_manager.py index e1fbef08..eaccb4cc 100644 --- a/src/base_odds_manager.py +++ b/src/base_odds_manager.py @@ -19,6 +19,7 @@ import json from typing import Dict, Any, Optional, List, cast from src.common.api_helper import DEFAULT_HTTP_HEADERS +from src.common.fetch_service import fetch_get, share_connection_pool @@ -59,7 +60,13 @@ class BaseOddsManager: # Deliberately no retry adapter, unlike api_helper: retries multiply # request_timeout, which is set to 5s precisely to stay inside that # budget. One try, then the cooldown below. + # + # Every scoreboard league manager builds one of these, so the session + # mounts the fetch service's shared no-retry adapter: the same single + # try, over one connection pool per host for all of them instead of + # one pool per instance. self.session = requests.Session() + share_connection_pool(self.session, max_retries=0) self.session.headers.update(DEFAULT_HTTP_HEADERS) # Configuration with defaults @@ -168,7 +175,7 @@ class BaseOddsManager: url = f"{self.base_url}/{sport}/leagues/{espn_league}/events/{event_id}/competitions/{event_id}/odds" self.logger.debug(f"Requesting odds from URL: {url}") - response = self.session.get(url, timeout=self.request_timeout) + response = fetch_get(self.session, url, timeout=self.request_timeout) response.raise_for_status() raw_data = response.json() diff --git a/src/common/README.md b/src/common/README.md index b707b685..9e9b7f6c 100644 --- a/src/common/README.md +++ b/src/common/README.md @@ -27,6 +27,7 @@ Rules for the package: | [`bdf_font`](#bdf_font) | Load and draw BDF bitmap fonts | Yes, if drawing BDF text directly | 3.5.0 | | [`espn_dates`](#espn_dates) | Fetch ESPN scoreboards across a date range | Yes (scoreboards) | 3.5.0 | | [`favorite_team_check`](#favorite_team_check) | Log why a favourite team code shows nothing | Yes (scoreboards) | 3.6.0 | +| [`fetch_service`](#fetch_service) | Pooled, merged, budgeted and counted HTTP for core fetch paths | No, core-internal (reached through `api_helper` and `espn_dates`) | n/a | | [`font_layout`](#font_layout) | Reproducible TrueType loading, crisp sizes | Yes | 3.4.0 | | [`frame_timing`](#frame_timing) | Timing of every presented frame, stall watchdog | No, core-internal | n/a | | [`json_body`](#json_body) | Parse a response body as JSON, with orjson if installed | Optional (large payloads) | 3.5.0 | @@ -108,7 +109,9 @@ and truncates results when `limit` is above 500. `fetch_espn_scoreboard()` splits a range into month and day requests ESPN accepts and merges the results; `espn_date_chunks()`, `fetch_espn_date_chunks()`, `clamp_espn_limit()` and `merge_scoreboard_payloads()` are the pieces. -Scoreboard plugins also bundle a copy for older cores. +Every request goes through [`fetch_service`](#fetch_service), the chunks +counted against the plugin that asked. Scoreboard plugins also bundle a copy +for older cores. ### favorite_team_check @@ -121,6 +124,23 @@ says the league has nothing on yet; `reset()` re-arms it after a config edit. Diagnostics only: every failure is swallowed. Scoreboard plugins also bundle a copy for older cores. +### fetch_service + +[`fetch_service.py`](fetch_service.py). Core-internal for now. Every core +fetch path -- `APIHelper.get`/`post`, `espn_dates` (so every scoreboard's +ESPN scoreboard fetch and `SportsFetchMixin`), `BackgroundDataService` and +`BaseOddsManager` -- calls `fetch_get(session, url, ...)` instead of +`session.get(url, ...)`. Same arguments, return value and exceptions; on top +it shares one connection pool per host per retry policy +(`share_connection_pool`), merges identical GETs in flight, applies per-host +token buckets (`fetch_service.rate_limits` in config.json; ESPN gets 20/s, +burst 200), revalidates with server-sent `ETag`/`Last-Modified` and counts +requests per plugin and per host. The display publishes the counters +(`FetchStatsPublisher`) for `GET /api/v3/plugins/fetch-stats`. Which plugin +made a request comes from `plugin_scope()`, set by the plugin executor, or +else from the plugin directory on the stack. See +[docs/PLUGIN_API_REFERENCE.md](../../docs/PLUGIN_API_REFERENCE.md#fetching-data). + ### font_layout [`font_layout.py`](font_layout.py). `load_truetype(path, size)` is diff --git a/src/common/api_helper.py b/src/common/api_helper.py index b5f07276..d9915f77 100644 --- a/src/common/api_helper.py +++ b/src/common/api_helper.py @@ -11,10 +11,10 @@ import time from datetime import datetime from types import MappingProxyType from src.common.espn_dates import ESPN_MAX_LIMIT +from src.common.fetch_service import fetch_get, fetch_post, share_connection_pool from typing import TYPE_CHECKING, Any, Dict, Mapping, Optional, cast import requests -from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry if TYPE_CHECKING: @@ -45,7 +45,11 @@ class APIHelper: - Requests go through one ``requests.Session`` that retries GET, HEAD and OPTIONS on 429 and 5xx with exponential backoff, and sends - :data:`DEFAULT_HTTP_HEADERS`. + :data:`DEFAULT_HTTP_HEADERS`. Its connection pool is shared with every + other helper using the same retry policy, and requests go through the + core fetch service (``src/common/fetch_service.py``): identical GETs in + flight are merged, hosts with a budget are paced, and requests are + counted per plugin. Return values and errors are unchanged. - Consecutive requests from one helper are spaced at least ``set_rate_limit()`` seconds apart (1 second by default). A cache hit does not count. @@ -81,9 +85,10 @@ class APIHelper: status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["GET", "HEAD", "OPTIONS"] ) - adapter = HTTPAdapter(max_retries=retry_strategy) - self.session.mount("https://", adapter) - self.session.mount("http://", adapter) + # The shared adapter for this retry policy: the same retries as a + # private HTTPAdapter(max_retries=retry_strategy), with the connection + # pool shared by every helper (fetch_service). + share_connection_pool(self.session, retry_strategy) self.session.headers.update({**DEFAULT_HTTP_HEADERS, 'Connection': 'keep-alive'}) @@ -128,7 +133,8 @@ class APIHelper: request_headers.update(headers) # Make request - response = self.session.get( + response = fetch_get( + self.session, url, params=params, headers=request_headers, @@ -255,7 +261,8 @@ class APIHelper: if headers: request_headers.update(headers) - response = self.session.post( + response = fetch_post( + self.session, url, data=data, json=json_data, diff --git a/src/common/espn_dates.py b/src/common/espn_dates.py index ac517677..66f859c8 100644 --- a/src/common/espn_dates.py +++ b/src/common/espn_dates.py @@ -32,6 +32,7 @@ scoreboards ask every 30 seconds. After that the range is tried again, so the workaround retires itself if ESPN reverts. """ +import contextvars import threading import time from concurrent.futures import ThreadPoolExecutor @@ -47,6 +48,21 @@ except ImportError: def response_json(response: Any) -> Any: return response.json() +try: + # The core fetch service: counts, per-host budget, merging of identical + # requests. Same call, same result and errors as ``session.get``. + from src.common.fetch_service import fetch_get, pinned_caller +except ImportError: + # Bundled copies on cores without it call the session directly. + import contextlib + + def fetch_get(session: Any, url: str, *, share_in_flight: bool = True, + **kwargs: Any) -> Any: + return session.get(url, **kwargs) + + def pinned_caller() -> Any: + return contextlib.nullcontext() + # Above this, ESPN returns a truncated list instead of an error. See module # docstring: 500 is the largest value measured to return complete data. ESPN_MAX_LIMIT = 500 @@ -195,7 +211,8 @@ def _fetch_one_chunk( logged and swallowed here rather than raised to the gather below. """ try: - response = session.get( + response = fetch_get( + session, url, params=dict(params, dates=chunk, limit=ESPN_MAX_LIMIT), headers=headers, @@ -220,6 +237,10 @@ def _fetch_chunks( callers keep ``chunks`` order from the returned list -- but it does mean the session is shared across threads, which is why this only ever issues GETs and never touches session state. + + Each chunk runs in a copy of the caller's context, with the caller pinned + into it, so the fetch service counts the chunks against the plugin that + asked for the range rather than against the core. """ if not chunks: return [] @@ -229,10 +250,15 @@ def _fetch_chunks( if len(chunks) == 1: return [fetch(chunks[0])] workers = min(ESPN_CHUNK_WORKERS, len(chunks)) + with pinned_caller(): + # One copy per chunk: a Context cannot be entered by two threads. + contexts = [contextvars.copy_context() for _ in chunks] with ThreadPoolExecutor( max_workers=workers, thread_name_prefix="espn-chunk", ) as pool: - return list(pool.map(fetch, chunks)) + futures = [pool.submit(context.run, fetch, chunk) + for context, chunk in zip(contexts, chunks)] + return [future.result() for future in futures] def fetch_espn_date_chunks( @@ -363,7 +389,7 @@ def fetch_espn_scoreboard( # real error to log, without spending the chunks a second time. chunks_tried = True - response = session.get(url, params=params, headers=headers, timeout=timeout) + response = fetch_get(session, url, params=params, headers=headers, timeout=timeout) if is_range and response.status_code == 400 and not chunks_tried: _note_range_rejected() if logger: diff --git a/src/common/fetch_service.py b/src/common/fetch_service.py new file mode 100644 index 00000000..0764b029 --- /dev/null +++ b/src/common/fetch_service.py @@ -0,0 +1,1173 @@ +"""One place every core HTTP fetch goes through: pooling, merging, budgets, counts. + +Stage 1 of the shared fetch service (docs/PLUGIN_API_REFERENCE.md, "Fetching +data"). Core's own fetch paths call :meth:`FetchService.get` instead of +``session.get``: + +- ``APIHelper.get`` / ``post`` (``src/common/api_helper.py``); +- ``fetch_espn_scoreboard`` and its date chunks (``src/common/espn_dates.py``), + which every scoreboard plugin's live, recent and upcoming fetch and + ``SportsFetchMixin._fetch_season_directly`` already use; +- ``BackgroundDataService`` (season schedules); +- ``BaseOddsManager.get_odds``. + +So the plugins that use those helpers are covered without changing a line of +plugin code. What the service adds, around the caller's own ``session.get``: + +**Connection pooling.** :func:`share_connection_pool` mounts one shared +``HTTPAdapter`` per retry policy on a caller's Session. urllib3 keeps one +connection pool per host inside it, so every core Session with the same retry +policy reuses the same TCP/TLS connections to a host -- the dozens of +``BaseOddsManager`` instances (one per scoreboard league manager) stop opening +a pool each. Each caller keeps its own Session object, so headers, cookies and +auth -- the mutable parts -- are never shared. :meth:`FetchService.session_for` +is the pooled per-host Session for a caller that has none. + +**Single-flight.** Identical GETs in flight at the same time go to the network +once; the others wait and receive a copy of that response (or the same +exception). Identical means the same URL and query, the same effective request +headers, the same timeout and the same retry policy, so a caller can never be +handed a result its own request could not have produced. + +**Per-host token buckets.** Each host may have a budget, ``per_second`` with a +``burst``. A request past the budget waits for a token, but never longer than +``max_wait_seconds``: past that it goes anyway, counted as an overrun. A +limiter that stalls a plugin's update budget is worse than one extra request. +The default budgets ESPN only (``*.espn.com``: 20/s, burst 200), far above the +steady-state rate (well under 1/s) and above a cold season fetch; anything else +is unthrottled. + +**Conditional GET.** When a 200 carries ``ETag`` or ``Last-Modified``, the +body is kept in a small bounded store and the next identical GET sends +``If-None-Match`` / ``If-Modified-Since``. A ``304`` is turned back into the +200 the caller would have got, body and all, so callers never see a 304 they +did not ask for. Only server-sent validators are used; a caller that sets its +own conditional headers gets the raw answer. (ESPN sent no validators when +this was written -- see the PR that added this module -- so on ESPN the store +stays empty and costs nothing.) + +**Counters.** Requests, merged requests, bytes, 304s, errors, HTTP errors, +adapter retries, throttled requests and seconds waited, per plugin and per +host. :class:`FetchStatsPublisher` publishes them for the web interface +(``GET /api/v3/plugins/fetch-stats``). + +CALLER IDENTITY +--------------- +Counters are kept per plugin without plugins saying who they are: + +1. A context variable, set by the core around the code it runs for a plugin: + ``PluginExecutor`` around ``update()``/``display()``, ``PluginManager`` + around loading a plugin (its constructor and ``on_enable``). The core + carries it across its own thread hops: ``BackgroundDataService`` records + the submitter at submit time and its worker runs under it, and + ``espn_dates`` copies it into each chunk thread. +2. When no scope is set -- a thread the plugin started itself, or a call the + display makes outside the executor -- the first stack frame whose source + file lives in a loaded plugin's directory names the plugin + (:func:`register_plugin_directory`). +3. Otherwise the request is the core's own (``"core"``). + +Nothing here raises on account of bookkeeping: a failure in counting, +keying or the validator store falls back to a plain ``session.get``. + +Core-internal for now (stage 1). Plugins reach it through ``APIHelper`` and +``espn_dates``; a plugin-facing API comes with stage 3. +""" + +from __future__ import annotations + +import contextlib +import contextvars +import copy +import http.cookiejar +import json +import logging +import math +import os +import sys +import threading +import time +from collections import OrderedDict +from dataclasses import dataclass, field +from typing import ( + Any, Callable, Dict, Iterator, List, Mapping, Optional, Tuple, +) +from urllib.parse import urlsplit + +import requests +from requests.adapters import HTTPAdapter +from requests.models import PreparedRequest +from requests.sessions import merge_setting +from requests.structures import CaseInsensitiveDict + +logger = logging.getLogger(__name__) + +__all__ = [ + "CORE", + "DEFAULT_CONFIG", + "FETCH_STATS_KEY", + "FetchService", + "FetchStatsPublisher", + "TokenBucket", + "configure_fetch_service", + "current_plugin_id", + "fetch_get", + "fetch_post", + "get_fetch_service", + "pinned_caller", + "plugin_scope", + "read_fetch_stats", + "register_plugin_directory", + "share_connection_pool", + "start_fetch_stats_publisher", + "unregister_plugin_directory", +] + +#: The caller name for requests no plugin made. +CORE = "core" + +#: Read when the display starts: ``fetch_service`` in config.json, merged over +#: these. ``rate_limits`` maps a host (``api.example.com``) or a suffix pattern +#: (``*.espn.com``, which also matches ``espn.com``) to a budget; a +#: ``per_second`` of 0 or null removes the budget for that host. +DEFAULT_CONFIG: Mapping[str, Any] = { + "enabled": True, + "single_flight": True, + "conditional_get": True, + "max_wait_seconds": 2.0, + "rate_limits": { + "*.espn.com": {"per_second": 20, "burst": 200}, + }, + "validator_store": { + "max_entries": 64, + "max_bytes": 4 * 1024 * 1024, + "max_entry_bytes": 1024 * 1024, + }, +} + +#: Connection pools kept per shared adapter (one per host) and connections +#: kept per pool. Larger than requests' 10 because one adapter now serves +#: every core Session with its retry policy: three background workers each +#: running six ESPN chunk threads is 18 concurrent requests to one host. +_POOL_CONNECTIONS = 16 +_POOL_MAXSIZE = 24 + +#: Distinct hosts tracked individually; the rest are counted under "other". +_MAX_HOSTS = 200 +_MAX_PLUGINS = 200 +_OTHER = "other" + +#: How far up the stack the plugin-directory lookup looks. +_MAX_STACK_DEPTH = 80 + +_COUNTER_FIELDS = ( + "requests", # round trips sent (adapter retries inside one are not extra) + "merged", # answered by an identical request already in flight + "not_modified", # 304s turned back into the stored 200 + "errors", # transport exceptions (timeouts, connection errors, ...) + "http_errors", # responses with status >= 400 + "retries", # retries the urllib3 adapter made inside a request + "throttled", # requests that waited for a host budget + "overruns", # requests that went after max_wait_seconds anyway + "bytes", # decoded response body bytes received + "wait_seconds", # time spent waiting for host budgets +) + + +# --- caller identity --------------------------------------------------------- + +_current_plugin: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar( + "ledmatrix_fetch_plugin", default=None) + +_dirs_lock = threading.Lock() +#: (normalised directory prefix ending in a separator, plugin id), longest +#: prefix first so a nested directory wins. +_plugin_dirs: Tuple[Tuple[str, str], ...] = () +#: co_filename -> plugin id or "" (not a plugin file). Cleared whenever the +#: registered directories change. +_file_owner: Dict[str, str] = {} + + +def _norm_dir(path: str) -> str: + text = os.path.normcase(os.path.abspath(path)) + return text if text.endswith(os.sep) else text + os.sep + + +def register_plugin_directory(plugin_id: str, path: Any) -> None: + """Attribute code under ``path`` to ``plugin_id`` (the stack fallback). + + Called by the plugin manager when it loads a plugin. Both the path as + given and its real path are registered: plugin modules import from + ``sys.path`` (the directory as given, possibly a dev symlink) and from + the resolved entry point. + """ + global _plugin_dirs + if not plugin_id or path is None: + return + prefixes = {_norm_dir(str(path))} + try: + prefixes.add(_norm_dir(os.path.realpath(str(path)))) + except OSError: + pass + with _dirs_lock: + entries = {prefix: pid for prefix, pid in _plugin_dirs if pid != plugin_id} + for prefix in prefixes: + entries[prefix] = plugin_id + _plugin_dirs = tuple(sorted(entries.items(), key=lambda item: -len(item[0]))) + _file_owner.clear() + + +def unregister_plugin_directory(plugin_id: str) -> None: + """Forget ``plugin_id``'s directories (the plugin was unloaded).""" + global _plugin_dirs + with _dirs_lock: + _plugin_dirs = tuple(item for item in _plugin_dirs if item[1] != plugin_id) + _file_owner.clear() + + +def _owner_of_file(filename: str, dirs: Tuple[Tuple[str, str], ...]) -> str: + owner = _file_owner.get(filename) + if owner is not None: + return owner + normalised = os.path.normcase(filename) + owner = "" + for prefix, plugin_id in dirs: + if normalised.startswith(prefix): + owner = plugin_id + break + if len(_file_owner) < 10000: + _file_owner[filename] = owner + return owner + + +def _plugin_from_stack() -> Optional[str]: + dirs = _plugin_dirs + if not dirs: + return None + frame: Any = sys._getframe(1) + depth = 0 + while frame is not None and depth < _MAX_STACK_DEPTH: + owner = _owner_of_file(frame.f_code.co_filename, dirs) + if owner: + return owner + frame = frame.f_back + depth += 1 + return None + + +def current_plugin_id() -> Optional[str]: + """The plugin the current code is running for, or None for the core.""" + plugin_id = _current_plugin.get() + if plugin_id: + return plugin_id + try: + return _plugin_from_stack() + except Exception: # never let attribution break a fetch + return None + + +@contextlib.contextmanager +def plugin_scope(plugin_id: Optional[str]) -> Iterator[None]: + """Attribute fetches made inside the block (on this thread, and on threads + the core starts from it) to ``plugin_id``. ``None`` leaves the current + attribution alone.""" + if not plugin_id: + yield + return + token = _current_plugin.set(plugin_id) + try: + yield + finally: + _current_plugin.reset(token) + + +@contextlib.contextmanager +def pinned_caller() -> Iterator[None]: + """Resolve the caller now and pin it into the context, so a context copied + inside the block (``contextvars.copy_context()``) for a worker thread + carries the plugin even when only the stack could name it.""" + with plugin_scope(current_plugin_id()): + yield + + +# --- token bucket ------------------------------------------------------------- + +class TokenBucket: + """``per_second`` tokens a second, holding at most ``burst``. + + :meth:`reserve` takes a token and says how long the caller must wait for + it. Tokens may go negative -- each caller reserves its own slot -- but + never further than ``max_wait`` seconds' worth, so no request waits + longer than that. + """ + + def __init__(self, per_second: float, burst: float, + clock: Callable[[], float] = time.monotonic) -> None: + self.per_second = float(per_second) + self.burst = max(1.0, float(burst)) + self._clock = clock + self._tokens = self.burst + self._last = clock() + self._lock = threading.Lock() + + def reserve(self, max_wait: float) -> Tuple[float, bool]: + """Take a token. Returns (seconds to wait, whether max_wait capped it).""" + with self._lock: + now = self._clock() + elapsed = max(0.0, now - self._last) + self._last = now + self._tokens = min(self.burst, self._tokens + elapsed * self.per_second) + self._tokens -= 1.0 + if self._tokens >= 0: + return 0.0, False + wait = -self._tokens / self.per_second + if wait > max_wait: + self._tokens = -max_wait * self.per_second + return max(0.0, max_wait), True + return wait, False + + @property + def tokens(self) -> float: + with self._lock: + return self._tokens + + +# --- validator store ------------------------------------------------------------ + +@dataclass +class _Stored: + etag: Optional[str] + last_modified: Optional[str] + body: bytes + headers: Dict[str, str] + encoding: Optional[str] + + +class _ValidatorStore: + """LRU of (body, validators) for responses that carried validators.""" + + def __init__(self, max_entries: int, max_bytes: int, max_entry_bytes: int) -> None: + self.max_entries = max_entries + self.max_bytes = max_bytes + self.max_entry_bytes = max_entry_bytes + self._entries: "OrderedDict[Any, _Stored]" = OrderedDict() + self._bytes = 0 + self._lock = threading.Lock() + + def get(self, key: Any) -> Optional[_Stored]: + with self._lock: + entry = self._entries.get(key) + if entry is not None: + self._entries.move_to_end(key) + return entry + + def put(self, key: Any, entry: _Stored) -> None: + size = len(entry.body) + with self._lock: + self._drop_locked(key) + if size > self.max_entry_bytes or self.max_entries <= 0: + return + self._entries[key] = entry + self._bytes += size + while self._entries and (len(self._entries) > self.max_entries + or self._bytes > self.max_bytes): + _, old = self._entries.popitem(last=False) + self._bytes -= len(old.body) + + def drop(self, key: Any) -> None: + with self._lock: + self._drop_locked(key) + + def _drop_locked(self, key: Any) -> None: + old = self._entries.pop(key, None) + if old is not None: + self._bytes -= len(old.body) + + def clear(self) -> None: + with self._lock: + self._entries.clear() + self._bytes = 0 + + def stats(self) -> Dict[str, int]: + with self._lock: + return {"entries": len(self._entries), "bytes": self._bytes} + + +# --- helpers ------------------------------------------------------------------------ + +def _host_of(url: Any) -> str: + try: + host = urlsplit(str(url)).hostname + except ValueError: + host = None + return (host or "unknown").lower() + + +def _prepared_url(url: str, params: Any) -> str: + prepared = PreparedRequest() + prepared.prepare_url(url, params) + return str(prepared.url) + + +def _retry_fingerprint(retries: Any) -> Tuple[Any, ...]: + """What makes two urllib3 Retry configurations behave alike.""" + names = ("total", "connect", "read", "redirect", "status", "other", + "backoff_factor", "backoff_max", "raise_on_redirect", + "raise_on_status", "respect_retry_after_header") + values: List[Any] = [type(retries).__name__] + for name in names: + values.append(getattr(retries, name, None)) + for name in ("status_forcelist", "allowed_methods"): + value = getattr(retries, name, None) + if isinstance(value, (set, frozenset, list, tuple)): + value = tuple(sorted(str(v) for v in value)) + values.append(value if value is None or isinstance(value, (tuple, bool)) else repr(value)) + return tuple(values) + + +def _adapter_fingerprint(adapter: Any) -> Tuple[Any, ...]: + if type(adapter) is HTTPAdapter: + return ("HTTPAdapter", _retry_fingerprint(adapter.max_retries)) + return ("adapter", id(adapter)) + + +def _header_items(headers: Any) -> Tuple[Tuple[str, str], ...]: + if not headers: + return () + return tuple(sorted((str(k).lower(), str(v)) for k, v in dict(headers).items() + if v is not None)) + + +def _has_conditional_headers(headers: Any) -> bool: + if not headers: + return False + try: + names = {str(k).lower() for k in dict(headers)} + except (TypeError, ValueError): + return True + return bool(names & {"if-none-match", "if-modified-since", "if-match", + "if-unmodified-since", "if-range"}) + + +def _str_header(response: Any, name: str) -> Optional[str]: + headers = getattr(response, "headers", None) + try: + value = headers.get(name) if headers is not None else None + except Exception: + return None + return value if isinstance(value, str) and value else None + + +def _status_of(response: Any) -> Optional[int]: + status = getattr(response, "status_code", None) + return status if isinstance(status, int) and not isinstance(status, bool) else None + + +def _body_of(response: Any) -> Optional[bytes]: + """The already-read body, or None. Never reads a stream.""" + if not isinstance(response, requests.Response): + return None + content = getattr(response, "_content", None) + return content if isinstance(content, bytes) else None + + +def _retries_of(response: Any) -> int: + raw = getattr(response, "raw", None) + retries = getattr(raw, "retries", None) + history = getattr(retries, "history", None) + return len(history) if isinstance(history, tuple) else 0 + + +def _clone_response(response: Any) -> Any: + """A copy of a finished response for a merged caller.""" + if not isinstance(response, requests.Response): + return response + clone = copy.copy(response) + clone.headers = CaseInsensitiveDict(response.headers) + return clone + + +def _clone_error(error: BaseException) -> BaseException: + try: + return copy.copy(error) + except Exception: + return error + + +def _from_stored(entry: _Stored, not_modified: requests.Response) -> requests.Response: + """The 200 the caller would have got, rebuilt from a 304 and the store.""" + response = requests.Response() + response.status_code = 200 + response.reason = "OK" + response._content = entry.body + response._content_consumed = True # type: ignore[attr-defined] + headers = CaseInsensitiveDict(entry.headers) + for name, value in (not_modified.headers or {}).items(): + if name.lower() not in ("content-length", "content-encoding", "transfer-encoding"): + headers[name] = value + response.headers = headers + response.encoding = entry.encoding + response.url = not_modified.url + response.request = not_modified.request + response.history = not_modified.history + response.elapsed = not_modified.elapsed + response.cookies = not_modified.cookies + response.connection = getattr(not_modified, "connection", None) # type: ignore[assignment] + response.raw = not_modified.raw + return response + + +def _as_float(value: Any, default: float) -> float: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return default + number = float(value) + return number if math.isfinite(number) and number >= 0 else default + + +def _as_int(value: Any, default: int) -> int: + if isinstance(value, bool) or not isinstance(value, int): + return default + return value if value >= 0 else default + + +@dataclass +class _Flight: + event: threading.Event = field(default_factory=threading.Event) + response: Any = None + error: Optional[BaseException] = None + #: Callers waiting on this flight (read by tests). + waiters: int = 0 + + +@dataclass +class _Request: + """What one call resolved to before it goes out.""" + plugin: str + host: str + # Same URL + effective headers: what the validator store is keyed by. + representation: Optional[Tuple[Any, ...]] = None + # representation + timeout + retry policy + other kwargs: what merges. + flight: Optional[Tuple[Any, ...]] = None + + +# --- the service ---------------------------------------------------------------------- + +class FetchService: + """See the module docstring. One per process: :func:`get_fetch_service`.""" + + def __init__(self, config: Any = None, *, + clock: Callable[[], float] = time.monotonic, + sleep: Callable[[float], None] = time.sleep, + wall_clock: Callable[[], float] = time.time) -> None: + self._clock = clock + self._sleep = sleep + self._wall_clock = wall_clock + self.started_at = wall_clock() + self._lock = threading.Lock() + self._inflight: Dict[Tuple[Any, ...], _Flight] = {} + self._buckets: Dict[str, Optional[TokenBucket]] = {} + self._adapters: Dict[Tuple[Any, ...], HTTPAdapter] = {} + self._sessions: Dict[str, requests.Session] = {} + self._plugins: Dict[str, Dict[str, float]] = {} + self._plugin_hosts: Dict[str, Dict[str, int]] = {} + self._hosts: Dict[str, Dict[str, float]] = {} + self._totals: Dict[str, float] = dict.fromkeys(_COUNTER_FIELDS, 0) + #: Bumped on every counter change; the publisher compares it. + self.change_count = 0 + self.enabled = True + self.single_flight = True + self.conditional_get = True + self.max_wait_seconds = 2.0 + self._rate_limits: Dict[str, Tuple[float, float]] = {} + self._validators = _ValidatorStore(0, 0, 0) + self._applied: Optional[str] = None + self.configure(config) + + # -- configuration -- + + def configure(self, config: Any = None) -> None: + """Apply ``config`` (the ``fetch_service`` section) over DEFAULT_CONFIG. + Bad values are ignored with a warning; this never raises. Applying + the same section again is a no-op: a config hot reload for any other + setting keeps the budgets and the validator store.""" + try: + signature = json.dumps(config, sort_keys=True, default=repr) + except (TypeError, ValueError): + signature = repr(config) + if signature == self._applied: + return + self._applied = signature + merged: Dict[str, Any] = dict(DEFAULT_CONFIG) + if isinstance(config, Mapping): + merged.update({k: v for k, v in config.items() if v is not None}) + elif config is not None: + logger.warning("fetch_service config is not an object; using the defaults") + + self.enabled = merged.get("enabled") is not False + self.single_flight = merged.get("single_flight") is not False + self.conditional_get = merged.get("conditional_get") is not False + self.max_wait_seconds = _as_float(merged.get("max_wait_seconds"), + float(DEFAULT_CONFIG["max_wait_seconds"])) + + limits: Dict[str, Tuple[float, float]] = {} + raw_limits = merged.get("rate_limits") + if isinstance(raw_limits, Mapping): + for pattern, spec in raw_limits.items(): + if not isinstance(pattern, str) or not pattern.strip(): + continue + if spec is None: + continue + if not isinstance(spec, Mapping): + logger.warning("fetch_service.rate_limits[%r] is not an object; ignored", pattern) + continue + per_second = _as_float(spec.get("per_second"), -1.0) + if spec.get("per_second") in (None, 0): + continue # explicitly unthrottled + if per_second <= 0: + logger.warning("fetch_service.rate_limits[%r].per_second must be a " + "positive number; ignored", pattern) + continue + burst = _as_float(spec.get("burst"), max(1.0, per_second)) + limits[pattern.strip().lower()] = (per_second, max(1.0, burst)) + elif raw_limits is not None: + logger.warning("fetch_service.rate_limits is not an object; no host budgets") + + store = merged.get("validator_store") + store = store if isinstance(store, Mapping) else {} + default_store = DEFAULT_CONFIG["validator_store"] + with self._lock: + self._rate_limits = limits + self._buckets.clear() + self._validators = _ValidatorStore( + _as_int(store.get("max_entries"), default_store["max_entries"]), + _as_int(store.get("max_bytes"), default_store["max_bytes"]), + _as_int(store.get("max_entry_bytes"), default_store["max_entry_bytes"]), + ) + self.change_count += 1 + + def describe_config(self) -> Dict[str, Any]: + with self._lock: + limits = {pattern: {"per_second": rate, "burst": burst} + for pattern, (rate, burst) in sorted(self._rate_limits.items())} + return { + "enabled": self.enabled, + "single_flight": self.single_flight, + "conditional_get": self.conditional_get, + "max_wait_seconds": self.max_wait_seconds, + "rate_limits": limits, + } + + def _limit_for(self, host: str) -> Optional[Tuple[float, float]]: + limits = self._rate_limits + exact = limits.get(host) + if exact is not None: + return exact + best: Optional[Tuple[float, float]] = None + best_len = -1 + for pattern, limit in limits.items(): + if pattern.startswith("*."): + suffix = pattern[1:] # ".espn.com" + if (host.endswith(suffix) or host == suffix[1:]) and len(suffix) > best_len: + best, best_len = limit, len(suffix) + return best + + def _bucket_for(self, host: str) -> Optional[TokenBucket]: + with self._lock: + if host in self._buckets: + return self._buckets[host] + limit = self._limit_for(host) + bucket = None if limit is None else TokenBucket(limit[0], limit[1], self._clock) + if len(self._buckets) < 1000: + self._buckets[host] = bucket + return bucket + + # -- pooling -- + + def shared_adapter(self, max_retries: Any = 0) -> HTTPAdapter: + """The shared adapter for a retry policy (an int or a urllib3 Retry).""" + probe = HTTPAdapter(max_retries=max_retries) + key = _retry_fingerprint(probe.max_retries) + with self._lock: + adapter = self._adapters.get(key) + if adapter is None: + adapter = HTTPAdapter(pool_connections=_POOL_CONNECTIONS, + pool_maxsize=_POOL_MAXSIZE, + max_retries=max_retries) + self._adapters[key] = adapter + return adapter + + def session_for(self, url: str) -> requests.Session: + """The pooled Session for ``url``'s host, for a caller without one. + + Stateless on purpose -- it is shared by every such caller -- so it + accepts no cookies; pass headers per request. + """ + host = _host_of(url) + with self._lock: + session = self._sessions.get(host) + if session is not None: + return session + session = requests.Session() + session.cookies.set_policy(http.cookiejar.DefaultCookiePolicy(allowed_domains=[])) + try: + from src.common.api_helper import DEFAULT_HTTP_HEADERS + session.headers.update(DEFAULT_HTTP_HEADERS) + except ImportError: # pragma: no cover - always importable in core + pass + adapter = self.shared_adapter(0) + session.mount("https://", adapter) + session.mount("http://", adapter) + with self._lock: + return self._sessions.setdefault(host, session) + + # -- requests -- + + def get(self, session: Any, url: str, *, share_in_flight: bool = True, + **kwargs: Any) -> Any: + """``session.get(url, **kwargs)`` through the service. + + Same return value, same exceptions, and ``session.get`` is called + with the caller's own arguments (plus validator headers when the + store holds some). ``session=None`` uses :meth:`session_for`. + + ``share_in_flight=False`` always sends this request itself rather + than joining an identical one in flight -- for a caller that may + retry *because* an earlier request hung (BackgroundDataService + cancels and replaces a fetch) and must not be handed that one. + """ + transport = session if session is not None else self.session_for(url) + if not self.enabled: + return transport.get(url, **kwargs) + try: + request = self._resolve(transport, url, kwargs) + except Exception: + logger.debug("fetch_service could not key a request to %s", url, exc_info=True) + request = _Request(plugin=self._caller(), host=_host_of(url)) + + if request.flight is None or not self.single_flight or not share_in_flight: + return self._send_get(transport, url, kwargs, request) + + with self._lock: + flight = self._inflight.get(request.flight) + leader = flight is None + if flight is None: + flight = _Flight() + self._inflight[request.flight] = flight + else: + flight.waiters += 1 + if not leader: + flight.event.wait() + self._count(request.plugin, request.host, merged=1) + if flight.error is not None: + raise _clone_error(flight.error) + return _clone_response(flight.response) + try: + response = self._send_get(transport, url, kwargs, request) + flight.response = response + return response + except BaseException as error: + flight.error = error + raise + finally: + with self._lock: + if self._inflight.get(request.flight) is flight: + del self._inflight[request.flight] + flight.event.set() + + def post(self, session: Any, url: str, **kwargs: Any) -> Any: + """``session.post(url, **kwargs)``: counted and budgeted, never merged.""" + transport = session if session is not None else self.session_for(url) + if not self.enabled: + return transport.post(url, **kwargs) + request = _Request(plugin=self._caller(), host=_host_of(url)) + return self._send(lambda: transport.post(url, **kwargs), request) + + def note_merged(self, url: Any = None, plugin_id: Optional[str] = None) -> None: + """Count a request answered by another caller's fetch outside this + service (BackgroundDataService joining an in-flight cache key).""" + try: + self._count(plugin_id or self._caller(), _host_of(url), merged=1) + except Exception: + logger.debug("fetch_service could not count a merged request", exc_info=True) + + def _caller(self) -> str: + return current_plugin_id() or CORE + + def _resolve(self, transport: Any, url: str, kwargs: Dict[str, Any]) -> _Request: + request = _Request(plugin=self._caller(), host=_host_of(url)) + if kwargs.get("stream"): + return request # body not read here: nothing to share or store + full_url = _prepared_url(url, kwargs.get("params")) + headers = kwargs.get("headers") + identity = getattr(transport, "fetch_identity_session", None) + session = identity if isinstance(identity, requests.Session) else transport + if isinstance(session, requests.Session): + effective = merge_setting(headers, session.headers, dict_class=CaseInsensitiveDict) + header_items = _header_items(effective) + policy: Tuple[Any, ...] = (type(transport).__name__, + _adapter_fingerprint(session.get_adapter(full_url))) + stateful = bool(session.cookies) or session.auth is not None + owner = id(session) if stateful else None + else: + header_items = _header_items(headers) + policy = (type(transport).__name__, id(transport)) + owner = id(transport) + request.representation = (full_url, header_items, owner) + others = tuple(sorted((k, repr(v)) for k, v in kwargs.items() + if k not in ("params", "headers"))) + request.flight = ("GET", request.representation, policy, others) + return request + + def _send_get(self, transport: Any, url: str, kwargs: Dict[str, Any], + request: _Request) -> Any: + stored: Optional[_Stored] = None + call_kwargs = kwargs + conditional = (self.conditional_get and request.representation is not None + and not _has_conditional_headers(kwargs.get("headers"))) + if conditional: + stored = self._validators.get(request.representation) + if stored is not None: + extra: Dict[str, str] = {} + if stored.etag: + extra["If-None-Match"] = stored.etag + if stored.last_modified: + extra["If-Modified-Since"] = stored.last_modified + headers = kwargs.get("headers") + merged_headers: Any = (CaseInsensitiveDict(headers) if headers + else CaseInsensitiveDict()) + merged_headers.update(extra) + call_kwargs = dict(kwargs, headers=merged_headers) + + response = self._send(lambda: transport.get(url, **call_kwargs), request, + stored=stored) + if conditional: + try: + response = self._after_conditional(request, response, stored) + except Exception: + logger.debug("fetch_service validator bookkeeping failed for %s", url, + exc_info=True) + return response + + def _after_conditional(self, request: _Request, response: Any, + stored: Optional[_Stored]) -> Any: + status = _status_of(response) + key = request.representation + if status == 304 and stored is not None and isinstance(response, requests.Response): + return _from_stored(stored, response) + if status == 200: + body = _body_of(response) + etag = _str_header(response, "ETag") + last_modified = _str_header(response, "Last-Modified") + no_store = "no-store" in (_str_header(response, "Cache-Control") or "").lower() + if body is not None and (etag or last_modified) and not no_store: + self._validators.put(key, _Stored( + etag=etag, last_modified=last_modified, body=body, + headers=dict(response.headers), encoding=response.encoding)) + else: + self._validators.drop(key) + return response + + def _send(self, call: Callable[[], Any], request: _Request, + stored: Optional[_Stored] = None) -> Any: + waited = 0.0 + overrun = False + try: + bucket = self._bucket_for(request.host) + if bucket is not None: + waited, overrun = bucket.reserve(self.max_wait_seconds) + except Exception: + logger.debug("fetch_service budget check failed", exc_info=True) + waited, overrun = 0.0, False + if waited > 0: + self._sleep(waited) + try: + response = call() + except BaseException: + self._count(request.plugin, request.host, requests=1, errors=1, + throttled=int(waited > 0), overruns=int(overrun), + wait_seconds=waited) + raise + try: + status = _status_of(response) + body = _body_of(response) + not_modified = int(status == 304 and stored is not None) + self._count(request.plugin, request.host, requests=1, + not_modified=not_modified, + http_errors=int(status is not None and status >= 400), + retries=_retries_of(response), + bytes=len(body) if body is not None else 0, + throttled=int(waited > 0), overruns=int(overrun), + wait_seconds=waited) + except Exception: + logger.debug("fetch_service could not count a response", exc_info=True) + return response + + # -- counters -- + + def _count(self, plugin: str, host: str, **deltas: float) -> None: + """Add ``deltas`` to the plugin's, the host's and the totals. Never + raises: it runs on every fetch's success and failure paths.""" + try: + self._apply_counts(plugin, host, deltas) + except Exception: + logger.debug("fetch_service could not update its counters", exc_info=True) + + def _apply_counts(self, plugin: str, host: str, deltas: Dict[str, float]) -> None: + changes = {k: v for k, v in deltas.items() if v} + if not changes: + return + with self._lock: + if plugin not in self._plugins and len(self._plugins) >= _MAX_PLUGINS: + plugin = _OTHER + if host not in self._hosts and len(self._hosts) >= _MAX_HOSTS: + host = _OTHER + per_plugin = self._plugins.setdefault(plugin, dict.fromkeys(_COUNTER_FIELDS, 0)) + per_host = self._hosts.setdefault(host, dict.fromkeys(_COUNTER_FIELDS, 0)) + for name, value in changes.items(): + per_plugin[name] += value + per_host[name] += value + self._totals[name] += value + if changes.get("requests") or changes.get("merged"): + hosts = self._plugin_hosts.setdefault(plugin, {}) + hosts[host] = hosts.get(host, 0) + int(changes.get("requests", 0) + + changes.get("merged", 0)) + self.change_count += 1 + + def reset_counters(self) -> None: + with self._lock: + self._plugins.clear() + self._plugin_hosts.clear() + self._hosts.clear() + self._totals = dict.fromkeys(_COUNTER_FIELDS, 0) + self.change_count += 1 + + def reset(self) -> None: + """Counters, validators, budgets and in-flight table (tests).""" + self.reset_counters() + with self._lock: + self._buckets.clear() + self._inflight.clear() + self._validators.clear() + + def snapshot(self) -> Dict[str, Any]: + """Counters since the service started, JSON-ready.""" + def rounded(counters: Mapping[str, float]) -> Dict[str, Any]: + out: Dict[str, Any] = {} + for name in _COUNTER_FIELDS: + value = counters.get(name, 0) + out[name] = round(float(value), 3) if name == "wait_seconds" else int(value) + return out + + with self._lock: + plugins = {pid: dict(rounded(c), hosts=dict(self._plugin_hosts.get(pid, {}))) + for pid, c in self._plugins.items()} + hosts = {host: rounded(c) for host, c in self._hosts.items()} + totals = rounded(self._totals) + change = self.change_count + return { + "since": self.started_at, + "change_count": change, + "totals": totals, + "plugins": plugins, + "hosts": hosts, + "validators": self._validators.stats(), + "config": self.describe_config(), + } + + +# --- process-wide instance --------------------------------------------------------- + +_service: Optional[FetchService] = None +_service_lock = threading.Lock() + + +def get_fetch_service() -> FetchService: + """The process's FetchService, created with the defaults on first use.""" + global _service + service = _service + if service is not None: + return service + with _service_lock: + if _service is None: + _service = FetchService() + return _service + + +def configure_fetch_service(config: Any) -> FetchService: + """Apply config.json's ``fetch_service`` section. Never raises.""" + service = get_fetch_service() + try: + service.configure(config) + except Exception: + logger.warning("Could not apply the fetch_service config; keeping the defaults", + exc_info=True) + return service + + +def fetch_get(session: Any, url: str, *, share_in_flight: bool = True, + **kwargs: Any) -> Any: + """``session.get(url, **kwargs)`` through the process's FetchService.""" + return get_fetch_service().get(session, url, share_in_flight=share_in_flight, + **kwargs) + + +def fetch_post(session: Any, url: str, **kwargs: Any) -> Any: + """``session.post(url, **kwargs)`` through the process's FetchService.""" + return get_fetch_service().post(session, url, **kwargs) + + +def share_connection_pool(session: Any, max_retries: Any = 0) -> None: + """Mount the shared adapter for ``max_retries`` on ``session``'s http and + https prefixes, replacing what was there. Pass the same retry policy the + Session would otherwise mount: only the pool becomes shared. A no-op for + anything that is not a requests Session (a test double).""" + if not isinstance(session, requests.Session): + return + adapter = get_fetch_service().shared_adapter(max_retries) + session.mount("https://", adapter) + session.mount("http://", adapter) + + +# --- publishing (display) and reading (web) ---------------------------------------- + +FETCH_STATS_KEY = "fetch_stats_snapshot" +SNAPSHOT_SCHEMA = 1 + +#: Shortest gap between two writes: the snapshot is on the SD card. +MIN_INTERVAL = 60.0 +#: An unchanged snapshot is rewritten this often so readers can tell a quiet +#: display from a dead one, and the cache's retention sweep keeps it. +REFRESH_INTERVAL = 600.0 +TICK_INTERVAL = 15.0 +STALE_AFTER = REFRESH_INTERVAL + 2 * MIN_INTERVAL + +LIVE = "live" +STALE = "stale" +STOPPED = "stopped" +UNKNOWN = "unknown" + + +class FetchStatsPublisher: + """Publishes the service's counters to the shared cache (display only). + + Written when the counters changed, at most once every ``min_interval`` + seconds, and otherwise every ``refresh_interval`` seconds. Never raises. + """ + + def __init__(self, cache_manager: Any, service: Optional[FetchService] = None, + min_interval: float = MIN_INTERVAL, + refresh_interval: float = REFRESH_INTERVAL, + clock: Callable[[], float] = time.monotonic, + wall_clock: Callable[[], float] = time.time) -> None: + self.cache_manager = cache_manager + self.service = service or get_fetch_service() + self.min_interval = min_interval + self.refresh_interval = refresh_interval + self._clock = clock + self._wall_clock = wall_clock + self._published_change: Optional[int] = None + self._last_attempt: Optional[float] = None + self._tick_lock = threading.Lock() + self._stop = threading.Event() + self._thread: Optional[threading.Thread] = None + + def _write(self, running: bool) -> None: + snapshot = self.service.snapshot() + snapshot.update({ + "schema": SNAPSHOT_SCHEMA, + "running": running, + "published_at": self._wall_clock(), + "stale_after": STALE_AFTER, + "pid": os.getpid(), + }) + self.cache_manager.set(FETCH_STATS_KEY, snapshot) + + def tick(self) -> bool: + """Publish if due. True if a snapshot was written.""" + with self._tick_lock: + try: + change = self.service.change_count + now = self._clock() + if self._last_attempt is not None: + since = now - self._last_attempt + if change == self._published_change: + if since < self.refresh_interval: + return False + elif since < self.min_interval: + return False + self._last_attempt = now + self._write(running=True) + self._published_change = change + return True + except Exception as err: + logger.debug("Could not publish fetch stats: %s", err, exc_info=True) + return False + + def start(self, interval: float = TICK_INTERVAL) -> None: + if self._thread is not None and self._thread.is_alive(): + return + self._stop.clear() + + def run() -> None: + # First write replaces whatever a previous run left behind. + self.tick() + while not self._stop.wait(interval): + self.tick() + + self._thread = threading.Thread(target=run, name="fetch-stats-publisher", daemon=True) + self._thread.start() + + def stop(self, publish_stopped: bool = True) -> None: + self._stop.set() + if self._thread is not None: + self._thread.join(timeout=2) + self._thread = None + if publish_stopped: + with self._tick_lock: + try: + self._write(running=False) + except Exception as err: + logger.debug("Could not publish stopped fetch stats: %s", err, exc_info=True) + + +def start_fetch_stats_publisher(cache_manager: Any) -> Optional[FetchStatsPublisher]: + """Start publishing this process's fetch counters. Display service only. + Never raises.""" + try: + publisher = FetchStatsPublisher(cache_manager) + publisher.start() + return publisher + except Exception as err: + logger.warning("Fetch statistics for the web interface are unavailable: %s", err) + return None + + +def read_fetch_stats(cache_manager: Any, now: Optional[float] = None) -> Dict[str, Any]: + """The display's latest counters, judged for staleness. Never raises. + + ``status`` is ``live``, ``stale`` (older than its ``stale_after``), + ``stopped`` (the display said so on exit; the final counters are still + given) or ``unknown`` (no readable snapshot). ``data`` is the snapshot, + or None when unknown. + """ + if cache_manager is None: + return {"status": UNKNOWN, "age_seconds": None, "data": None} + try: + snapshot = cache_manager.get(FETCH_STATS_KEY, max_age=None, memory_ttl=0) + except Exception as err: + logger.debug("Could not read fetch stats: %s", err, exc_info=True) + snapshot = None + if not isinstance(snapshot, dict) or snapshot.get("schema") != SNAPSHOT_SCHEMA: + return {"status": UNKNOWN, "age_seconds": None, "data": None} + published_at = snapshot.get("published_at") + if isinstance(published_at, bool) or not isinstance(published_at, (int, float)): + return {"status": UNKNOWN, "age_seconds": None, "data": None} + stale_after = _as_float(snapshot.get("stale_after"), STALE_AFTER) or STALE_AFTER + age = (time.time() if now is None else now) - float(published_at) + if snapshot.get("running") is not True: + status = STOPPED + elif age > stale_after or age < -stale_after: + status = STALE + else: + status = LIVE + return {"status": status, "age_seconds": round(age, 1), "data": snapshot} diff --git a/src/core_config_keys.py b/src/core_config_keys.py index 30eba5f5..42e75969 100644 --- a/src/core_config_keys.py +++ b/src/core_config_keys.py @@ -29,6 +29,7 @@ CORE_CONFIG_KEYS = frozenset({ 'display', 'sync', 'plugin_system', + 'fetch_service', # Older or optional core sections still found in existing config files. 'logging', 'network', diff --git a/src/display_controller.py b/src/display_controller.py index 0c35fa11..490aed4c 100644 --- a/src/display_controller.py +++ b/src/display_controller.py @@ -132,6 +132,13 @@ class DisplayController: # The web interface's /api/v3/errors/* read what this publishes. from src.error_aggregator import start_error_snapshot_publisher start_error_snapshot_publisher(self.cache_manager) + # Host budgets and the other fetch_service settings, before any plugin + # fetches; the web UI's fetch statistics read what the publisher + # writes (src/common/fetch_service.py). + from src.common.fetch_service import ( + configure_fetch_service, start_fetch_stats_publisher) + configure_fetch_service(self.config.get('fetch_service')) + self._fetch_stats_publisher = start_fetch_stats_publisher(self.cache_manager) logger.info("Config loaded in %.3f seconds (hot-reload: %s)", time.time() - start_time, enable_hot_reload) # Validate startup configuration. Errors are logged, not fatal. The @@ -3819,6 +3826,9 @@ class DisplayController: read stale values after the user saves settings via the web UI. """ self.config = new_config + # A no-op unless the fetch_service section itself changed. + from src.common.fetch_service import configure_fetch_service + configure_fetch_service(new_config.get('fetch_service')) self._normal_brightness = ( self.config.get('display', {}).get('hardware', {}).get('brightness', 90) ) @@ -3886,6 +3896,11 @@ class DisplayController: logger.warning("Error shutting down config service: %s", e) if getattr(self, '_font_usage_publisher', None) is not None: self._font_usage_publisher.stop() + if getattr(self, '_fetch_stats_publisher', None) is not None: + try: + self._fetch_stats_publisher.stop() + except Exception as e: + logger.warning("Error stopping the fetch statistics publisher: %s", e) # Publishes "stopped", so the web UI stops reporting what was loaded. if getattr(self, '_plugin_runtime_publisher', None) is not None: try: diff --git a/src/plugin_system/plugin_executor.py b/src/plugin_system/plugin_executor.py index c0e280ad..7c811f6f 100644 --- a/src/plugin_system/plugin_executor.py +++ b/src/plugin_system/plugin_executor.py @@ -10,6 +10,7 @@ from typing import Any, Dict, Optional, Callable from threading import Thread import logging +from src.common.fetch_service import plugin_scope from src.exceptions import PluginError from src.logging_config import get_logger from src.error_aggregator import record_error @@ -83,7 +84,10 @@ class PluginExecutor: def target(): try: - result_container['value'] = operation() + # Fetches made by the operation (and by threads the core + # starts from it) are counted against this plugin. + with plugin_scope(plugin_id): + result_container['value'] = operation() result_container['completed'] = True except Exception as e: result_container['exception'] = e diff --git a/src/plugin_system/plugin_manager.py b/src/plugin_system/plugin_manager.py index c59ce065..25b9b8f5 100644 --- a/src/plugin_system/plugin_manager.py +++ b/src/plugin_system/plugin_manager.py @@ -32,6 +32,7 @@ from src.plugin_system.schema_manager import ( from src.plugin_system.plugin_dirs import ( ManifestStatus, PluginDirectoryIndex, resolve_plugin_dir, ) +from src.common.fetch_service import plugin_scope, register_plugin_directory from src.common.permission_utils import ( ensure_directory_permissions, get_plugin_dir_mode @@ -423,6 +424,11 @@ class PluginManager: # Update mapping if found via search if plugin_id not in self.plugin_directories: self.plugin_directories[plugin_id] = plugin_dir + + # Code under this directory is this plugin's: the fetch service + # counts a request against it even from a thread the plugin + # started itself (src/common/fetch_service.py, caller identity). + register_plugin_directory(plugin_id, plugin_dir) # Get plugin config if self.config_manager: @@ -462,18 +468,20 @@ class PluginManager: config = dict(config) config['enabled'] = True - # Use PluginLoader to load plugin - plugin_instance, _module = self.plugin_loader.load_plugin( - plugin_id=plugin_id, - manifest=manifest, - plugin_dir=plugin_dir, - config=config, - display_manager=self.display_manager, - cache_manager=self.cache_manager, - plugin_manager=self, - install_deps=True, - plugins_dir=self.plugins_dir, - ) + # Use PluginLoader to load plugin. Fetches the constructor makes + # count against the plugin. + with plugin_scope(plugin_id): + plugin_instance, _module = self.plugin_loader.load_plugin( + plugin_id=plugin_id, + manifest=manifest, + plugin_dir=plugin_dir, + config=config, + display_manager=self.display_manager, + cache_manager=self.cache_manager, + plugin_manager=self, + install_deps=True, + plugins_dir=self.plugins_dir, + ) # Register plugin-shipped fonts with the FontManager (if any). # Plugin manifests can declare a "fonts" block that ships custom @@ -527,7 +535,8 @@ class PluginManager: # Call on_enable if plugin is enabled if hasattr(plugin_instance, 'on_enable'): try: - plugin_instance.on_enable() + with plugin_scope(plugin_id): + plugin_instance.on_enable() except Exception: # Undo the registration above before the outer # handler marks it ERROR: left in self.plugins, the diff --git a/test/fixtures/api_v3_url_map.json b/test/fixtures/api_v3_url_map.json index 52963d9c..707c8774 100644 --- a/test/fixtures/api_v3_url_map.json +++ b/test/fixtures/api_v3_url_map.json @@ -467,6 +467,15 @@ "POST" ] ], + [ + "/api/v3/plugins/fetch-stats", + "api_v3.get_fetch_stats", + [ + "GET", + "HEAD", + "OPTIONS" + ] + ], [ "/api/v3/plugins/health", "api_v3.get_plugin_health", diff --git a/test/test_fetch_service.py b/test/test_fetch_service.py new file mode 100644 index 00000000..397d9af9 --- /dev/null +++ b/test/test_fetch_service.py @@ -0,0 +1,878 @@ +"""The core fetch service: merging, host budgets, conditional GET, counters. + +No network: every request goes to a fake transport -- a real +``requests.Session`` subclass whose ``get`` answers from a handler -- so the +service sees real ``requests.Response`` objects, real header merging and real +adapters, and nothing leaves the machine. Clocks and sleeps are injected. + +What callers already rely on (return values, exceptions, retries) is pinned +by the existing suites, which run unchanged through the service: +test_api_helper.py, test_background_data_service*.py, +test_background_fetch_dedupe.py, test_base_odds_manager.py, +test_odds_request_budget.py, test_espn_dates.py and test_sports_fetch.py. +""" + +import importlib.util +import json +import threading +import time + +import pytest +import requests +from requests.structures import CaseInsensitiveDict +from urllib3.util.retry import Retry + +from src.common import fetch_service as fs +from src.common.fetch_service import ( + FetchService, + FetchStatsPublisher, + TokenBucket, + current_plugin_id, + plugin_scope, + read_fetch_stats, + register_plugin_directory, + unregister_plugin_directory, +) + + +# --- fakes ----------------------------------------------------------------------- + +class FakeClock: + def __init__(self, start=1000.0): + self.t = start + self.sleeps = [] + + def now(self): + return self.t + + def sleep(self, seconds): + self.sleeps.append(seconds) + self.t += seconds + + def advance(self, seconds): + self.t += seconds + + +def make_response(status=200, body=b'{"ok": 1}', headers=None, url="https://api.test/x"): + response = requests.Response() + response.status_code = status + response._content = body + response.headers = CaseInsensitiveDict(headers or {}) + response.url = url + response.encoding = "utf-8" + response.reason = "OK" if status < 400 else "Error" + return response + + +class FakeSession(requests.Session): + """A Session whose get() answers from ``handler(url, kwargs)``.""" + + def __init__(self, handler=None, gate=None): + super().__init__() + self.handler = handler or (lambda url, kwargs: make_response(url=url)) + self.gate = gate + self.calls = [] + self.started = threading.Event() + self._calls_lock = threading.Lock() + + def get(self, url, **kwargs): + with self._calls_lock: + self.calls.append((url, kwargs)) + self.started.set() + if self.gate is not None: + assert self.gate.wait(5), "test gate never opened" + return self.handler(url, kwargs) + + +@pytest.fixture +def clock(): + return FakeClock() + + +@pytest.fixture +def service(clock): + return FetchService({"rate_limits": {}}, clock=clock.now, sleep=clock.sleep) + + +@pytest.fixture +def global_service(monkeypatch, clock): + """A fresh process-wide service, for code that calls fetch_get().""" + svc = FetchService({"rate_limits": {}}, clock=clock.now, sleep=clock.sleep) + monkeypatch.setattr(fs, "_service", svc) + return svc + + +def _counters(svc, plugin=None, host=None): + snap = svc.snapshot() + if plugin is not None: + return snap["plugins"].get(plugin, {}) + if host is not None: + return snap["hosts"].get(host, {}) + return snap["totals"] + + +# --- the call itself is unchanged ----------------------------------------------------- + +class TestPassThrough: + + def test_session_get_sees_exactly_the_callers_arguments(self, service): + session = FakeSession() + response = service.get(session, "https://api.test/x", params={"a": 1}, + headers={"X-Y": "z"}, timeout=7) + assert session.calls == [("https://api.test/x", + {"params": {"a": 1}, "headers": {"X-Y": "z"}, "timeout": 7})] + assert response.json() == {"ok": 1} + + def test_no_kwargs_the_caller_did_not_pass(self, service): + session = FakeSession() + service.get(session, "https://api.test/x", timeout=5) + assert session.calls[0][1] == {"timeout": 5} + + def test_the_transport_exception_reaches_the_caller_unchanged(self, service): + boom = requests.ConnectionError("down") + + def handler(url, kwargs): + raise boom + + with pytest.raises(requests.ConnectionError) as caught: + service.get(FakeSession(handler), "https://api.test/x") + assert caught.value is boom + + def test_an_http_error_response_is_returned_not_raised(self, service): + session = FakeSession(lambda url, kw: make_response(503, b"busy")) + response = service.get(session, "https://api.test/x") + assert response.status_code == 503 + with pytest.raises(requests.HTTPError): + response.raise_for_status() + + def test_disabled_is_a_plain_session_get(self, clock): + svc = FetchService({"enabled": False, "rate_limits": {"api.test": {"per_second": 1, "burst": 1}}}, + clock=clock.now, sleep=clock.sleep) + session = FakeSession() + for _ in range(3): + svc.get(session, "https://api.test/x") + assert len(session.calls) == 3 + assert clock.sleeps == [] + assert _counters(svc)["requests"] == 0 + + def test_a_test_double_session_still_works(self, service): + from unittest.mock import MagicMock + session = MagicMock() + session.get.return_value.json.return_value = {"a": 1} + assert service.get(session, "https://api.test/x", timeout=3).json() == {"a": 1} + session.get.assert_called_once_with("https://api.test/x", timeout=3) + + def test_session_none_uses_the_pooled_session_for_the_host(self, service, monkeypatch): + seen = [] + monkeypatch.setattr(requests.Session, "get", + lambda self, url, **kw: seen.append(self) or make_response()) + service.get(None, "https://a.test/1") + service.get(None, "https://a.test/2") + service.get(None, "https://b.test/1") + assert seen[0] is seen[1] is service.session_for("https://a.test/") + assert seen[2] is not seen[0] + + +# --- single-flight ------------------------------------------------------------------------ + +def _wait_for_waiters(svc, count, timeout=5): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + with svc._lock: + flights = list(svc._inflight.values()) + if flights and flights[0].waiters >= count: + return + time.sleep(0.005) + raise AssertionError(f"only {flights[0].waiters if flights else 0} of {count} callers joined") + + +class TestSingleFlight: + + def test_concurrent_identical_gets_go_out_once(self, service): + gate = threading.Event() + session = FakeSession(gate=gate) + results = [] + + def call(): + results.append(service.get(session, "https://api.test/x", + params={"d": "1"}, timeout=5)) + + threads = [threading.Thread(target=call) for _ in range(5)] + threads[0].start() + assert session.started.wait(5) + for t in threads[1:]: + t.start() + _wait_for_waiters(service, 4) + gate.set() + for t in threads: + t.join(5) + + assert len(session.calls) == 1 + assert len(results) == 5 + assert all(r.json() == {"ok": 1} for r in results) + # Each caller gets its own Response object to mutate. + assert len({id(r) for r in results}) == 5 + totals = _counters(service) + assert totals["requests"] == 1 + assert totals["merged"] == 4 + + def test_merged_callers_get_the_leaders_exception(self, service): + gate = threading.Event() + + def handler(url, kwargs): + raise requests.Timeout("slow") + + session = FakeSession(handler, gate=gate) + errors = [] + + def call(): + try: + service.get(session, "https://api.test/x", timeout=5) + except requests.Timeout as err: + errors.append(err) + + threads = [threading.Thread(target=call) for _ in range(3)] + threads[0].start() + assert session.started.wait(5) + for t in threads[1:]: + t.start() + _wait_for_waiters(service, 2) + gate.set() + for t in threads: + t.join(5) + + assert len(session.calls) == 1 + assert len(errors) == 3 + totals = _counters(service) + assert totals["errors"] == 1 and totals["merged"] == 2 + + @pytest.mark.parametrize("second", [ + {"params": {"d": "2"}}, # another query + {"params": {"d": "1"}, "timeout": 9}, # another timeout + {"params": {"d": "1"}, "headers": {"Accept": "text/plain"}}, # another representation + ]) + def test_requests_that_could_answer_differently_are_not_merged(self, service, second): + gate = threading.Event() + session = FakeSession(gate=gate) + first = threading.Thread(target=lambda: service.get( + session, "https://api.test/x", params={"d": "1"}, timeout=5)) + first.start() + assert session.started.wait(5) + other = threading.Thread(target=lambda: service.get( + session, "https://api.test/x", **{"timeout": 5, **second})) + other.start() + deadline = time.monotonic() + 5 + while len(session.calls) < 2 and time.monotonic() < deadline: + time.sleep(0.005) + gate.set() + first.join(5) + other.join(5) + assert len(session.calls) == 2 + assert _counters(service)["merged"] == 0 + + def test_different_retry_policies_are_not_merged(self, service): + gate = threading.Event() + retrying = FakeSession(gate=gate) + retrying.mount("https://", requests.adapters.HTTPAdapter(max_retries=Retry(total=3))) + plain = FakeSession(gate=gate) + a = threading.Thread(target=lambda: service.get(retrying, "https://api.test/x")) + a.start() + assert retrying.started.wait(5) + b = threading.Thread(target=lambda: service.get(plain, "https://api.test/x")) + b.start() + assert plain.started.wait(5) + gate.set() + a.join(5) + b.join(5) + assert len(retrying.calls) == len(plain.calls) == 1 + + def test_sessions_with_the_same_policy_and_headers_share_a_flight(self, service): + gate = threading.Event() + one, two = FakeSession(gate=gate), FakeSession(gate=gate) + a = threading.Thread(target=lambda: service.get(one, "https://api.test/x", timeout=5)) + a.start() + assert one.started.wait(5) + b = threading.Thread(target=lambda: service.get(two, "https://api.test/x", timeout=5)) + b.start() + _wait_for_waiters(service, 1) + gate.set() + a.join(5) + b.join(5) + assert len(one.calls) == 1 and two.calls == [] + + def test_a_session_with_cookies_only_merges_with_itself(self, service): + gate = threading.Event() + cookied, plain = FakeSession(gate=gate), FakeSession(gate=gate) + cookied.cookies.set("sid", "secret") + a = threading.Thread(target=lambda: service.get(cookied, "https://api.test/x")) + a.start() + assert cookied.started.wait(5) + b = threading.Thread(target=lambda: service.get(plain, "https://api.test/x")) + b.start() + assert plain.started.wait(5) + gate.set() + a.join(5) + b.join(5) + assert len(cookied.calls) == len(plain.calls) == 1 + + def test_sequential_identical_gets_each_go_out(self, service): + session = FakeSession() + service.get(session, "https://api.test/x") + service.get(session, "https://api.test/x") + assert len(session.calls) == 2 + + def test_streamed_requests_are_never_merged(self, service): + gate = threading.Event() + session = FakeSession(gate=gate) + a = threading.Thread(target=lambda: service.get(session, "https://api.test/x", stream=True)) + a.start() + assert session.started.wait(5) + b = threading.Thread(target=lambda: service.get(session, "https://api.test/x", stream=True)) + b.start() + deadline = time.monotonic() + 5 + while len(session.calls) < 2 and time.monotonic() < deadline: + time.sleep(0.005) + gate.set() + a.join(5) + b.join(5) + assert len(session.calls) == 2 + + +# --- token buckets --------------------------------------------------------------------------- + +class TestTokenBucket: + + def test_burst_then_one_token_per_interval(self, clock): + bucket = TokenBucket(per_second=2, burst=3, clock=clock.now) + assert [bucket.reserve(10)[0] for _ in range(3)] == [0.0, 0.0, 0.0] + assert bucket.reserve(10) == (0.5, False) + assert bucket.reserve(10) == (1.0, False) + + def test_tokens_refill_with_time_up_to_the_burst(self, clock): + bucket = TokenBucket(per_second=2, burst=3, clock=clock.now) + for _ in range(3): + bucket.reserve(10) + clock.advance(100) + assert [bucket.reserve(10)[0] for _ in range(3)] == [0.0, 0.0, 0.0] + assert bucket.reserve(10)[0] == 0.5 + + def test_a_wait_is_capped_at_max_wait(self, clock): + bucket = TokenBucket(per_second=1, burst=1, clock=clock.now) + bucket.reserve(0.5) + assert bucket.reserve(0.5) == (0.5, True) + # The debt never runs further than max_wait either. + assert bucket.reserve(0.5) == (0.5, True) + clock.advance(10) + assert bucket.reserve(0.5) == (0.0, False) + + +class TestHostBudgets: + + def test_requests_past_the_budget_wait(self, clock): + svc = FetchService({"rate_limits": {"api.test": {"per_second": 1, "burst": 2}}, + "max_wait_seconds": 10}, clock=clock.now, sleep=clock.sleep) + session = FakeSession() + for _ in range(4): + svc.get(session, "https://api.test/x") + assert clock.sleeps == [1.0, 1.0] + host = _counters(svc, host="api.test") + assert host["throttled"] == 2 and host["wait_seconds"] == 2.0 + assert host["requests"] == 4 + + def test_other_hosts_are_not_throttled(self, clock): + svc = FetchService({"rate_limits": {"api.test": {"per_second": 1, "burst": 1}}}, + clock=clock.now, sleep=clock.sleep) + session = FakeSession() + for _ in range(5): + svc.get(session, "https://elsewhere.test/x") + assert clock.sleeps == [] + + def test_each_host_has_its_own_bucket(self, clock): + svc = FetchService({"rate_limits": {"*.espn.com": {"per_second": 1, "burst": 1}}, + "max_wait_seconds": 10}, clock=clock.now, sleep=clock.sleep) + session = FakeSession() + svc.get(session, "https://site.api.espn.com/a") + svc.get(session, "https://sports.core.api.espn.com/a") + assert clock.sleeps == [] + svc.get(session, "https://site.api.espn.com/a") + assert clock.sleeps == [1.0] + + def test_wildcard_matches_the_bare_domain_and_subdomains_only(self, clock): + svc = FetchService({"rate_limits": {"*.espn.com": {"per_second": 5, "burst": 9}}}, + clock=clock.now, sleep=clock.sleep) + assert svc._limit_for("espn.com") == (5.0, 9.0) + assert svc._limit_for("site.api.espn.com") == (5.0, 9.0) + assert svc._limit_for("notespn.com") is None + + def test_the_default_budget_covers_espn_and_a_cold_season_burst(self, clock): + svc = FetchService(clock=clock.now, sleep=clock.sleep) + session = FakeSession() + for _ in range(200): + svc.get(session, "https://site.api.espn.com/x") + assert clock.sleeps == [] + svc.get(session, "https://site.api.espn.com/x") + assert clock.sleeps == [pytest.approx(0.05)] + svc.get(session, "https://api.example.org/x") + assert len(clock.sleeps) == 1 + + def test_zero_per_second_removes_a_budget(self, clock): + svc = FetchService({"rate_limits": {"*.espn.com": {"per_second": 0, "burst": 1}}}, + clock=clock.now, sleep=clock.sleep) + session = FakeSession() + for _ in range(5): + svc.get(session, "https://site.api.espn.com/x") + assert clock.sleeps == [] + + def test_a_merged_caller_spends_no_token(self, clock): + svc = FetchService({"rate_limits": {"api.test": {"per_second": 1, "burst": 1}}, + "max_wait_seconds": 10}, clock=clock.now, sleep=clock.sleep) + gate = threading.Event() + session = FakeSession(gate=gate) + a = threading.Thread(target=lambda: svc.get(session, "https://api.test/x")) + a.start() + assert session.started.wait(5) + b = threading.Thread(target=lambda: svc.get(session, "https://api.test/x")) + b.start() + _wait_for_waiters(svc, 1) + gate.set() + a.join(5) + b.join(5) + assert clock.sleeps == [] + + +# --- conditional GET ---------------------------------------------------------------------------- + +class Versioned: + """A server with one resource and an ETag, honouring If-None-Match.""" + + def __init__(self, validator="etag"): + self.version = 1 + self.validator = validator + self.seen = [] + + def body(self): + return json.dumps({"version": self.version}).encode() + + def tag(self): + if self.validator == "etag": + return {"ETag": f'"v{self.version}"'} + return {"Last-Modified": f"Thu, 01 Oct 2026 00:00:0{self.version} GMT"} + + def __call__(self, url, kwargs): + headers = CaseInsensitiveDict(kwargs.get("headers") or {}) + self.seen.append(dict(headers)) + current = self.tag() + if (headers.get("If-None-Match") == current.get("ETag") and "ETag" in current) or \ + (headers.get("If-Modified-Since") == current.get("Last-Modified") + and "Last-Modified" in current): + return make_response(304, b"", headers={**current, "Date": "now"}, url=url) + return make_response(200, self.body(), + headers={**current, "Content-Type": "application/json"}, url=url) + + +class TestConditionalGet: + + @pytest.mark.parametrize("validator,header", [("etag", "If-None-Match"), + ("last-modified", "If-Modified-Since")]) + def test_a_304_returns_the_stored_body_as_a_200(self, service, validator, header): + server = Versioned(validator) + session = FakeSession(server) + first = service.get(session, "https://api.test/x", timeout=5) + second = service.get(session, "https://api.test/x", timeout=5) + + assert header not in server.seen[0] + assert header in server.seen[1] + assert second.status_code == 200 + assert second.json() == first.json() == {"version": 1} + assert second.headers["Content-Type"] == "application/json" + second.raise_for_status() + totals = _counters(service) + assert totals["requests"] == 2 + assert totals["not_modified"] == 1 + assert totals["bytes"] == len(server.body()) # the 304 carried none + + def test_a_changed_resource_is_fetched_and_stored_again(self, service): + server = Versioned() + session = FakeSession(server) + service.get(session, "https://api.test/x") + server.version = 2 + changed = service.get(session, "https://api.test/x") + assert changed.json() == {"version": 2} + again = service.get(session, "https://api.test/x") + assert again.json() == {"version": 2} + assert server.seen[2]["If-None-Match"] == '"v2"' + + def test_no_validators_no_conditional_request(self, service): + session = FakeSession() # answers 200 with no ETag/Last-Modified + service.get(session, "https://api.test/x", headers={"A": "1"}) + service.get(session, "https://api.test/x", headers={"A": "1"}) + assert session.calls[1][1] == {"headers": {"A": "1"}} + assert service.snapshot()["validators"]["entries"] == 0 + + def test_a_200_without_validators_drops_the_stored_one(self, service): + server = Versioned() + session = FakeSession(server) + service.get(session, "https://api.test/x") + session.handler = lambda url, kw: make_response(200, b'{"new": 1}', url=url) + service.get(session, "https://api.test/x") + assert service.snapshot()["validators"]["entries"] == 0 + + def test_a_callers_own_conditional_request_is_left_alone(self, service): + server = Versioned() + session = FakeSession(server) + service.get(session, "https://api.test/x") + raw = service.get(session, "https://api.test/x", headers={"If-None-Match": '"v1"'}) + assert raw.status_code == 304 + + def test_validators_are_per_representation(self, service): + server = Versioned() + session = FakeSession(server) + service.get(session, "https://api.test/x", params={"d": "1"}) + service.get(session, "https://api.test/x", params={"d": "2"}) + assert "If-None-Match" not in server.seen[1] + + def test_a_body_too_big_for_the_store_is_not_kept(self, clock): + svc = FetchService({"rate_limits": {}, "validator_store": {"max_entry_bytes": 4}}, + clock=clock.now, sleep=clock.sleep) + server = Versioned() + session = FakeSession(server) + svc.get(session, "https://api.test/x") + svc.get(session, "https://api.test/x") + assert "If-None-Match" not in server.seen[1] + + def test_the_store_evicts_least_recently_used_past_its_budget(self, clock): + svc = FetchService({"rate_limits": {}, "validator_store": {"max_entries": 2}}, + clock=clock.now, sleep=clock.sleep) + session = FakeSession(Versioned()) + for path in ("a", "b", "c"): + svc.get(session, f"https://api.test/{path}") + assert svc.snapshot()["validators"]["entries"] == 2 + + def test_off_switch(self, clock): + svc = FetchService({"rate_limits": {}, "conditional_get": False}, + clock=clock.now, sleep=clock.sleep) + server = Versioned() + session = FakeSession(server) + svc.get(session, "https://api.test/x") + svc.get(session, "https://api.test/x") + assert "If-None-Match" not in server.seen[1] + + +# --- counters and caller identity ---------------------------------------------------------------- + +class TestCounters: + + def test_per_plugin_and_per_host(self, service): + session = FakeSession() + with plugin_scope("weather"): + service.get(session, "https://api.weather.test/now") + service.get(session, "https://api.weather.test/later") + service.get(session, "https://site.api.espn.com/x") + snap = service.snapshot() + assert snap["plugins"]["weather"]["requests"] == 2 + assert snap["plugins"]["weather"]["hosts"] == {"api.weather.test": 2} + assert snap["plugins"]["core"]["requests"] == 1 + assert snap["hosts"]["api.weather.test"]["requests"] == 2 + assert snap["hosts"]["site.api.espn.com"]["requests"] == 1 + assert snap["totals"]["bytes"] == 3 * len(b'{"ok": 1}') + + def test_errors_and_http_errors(self, service): + def handler(url, kwargs): + if url.endswith("/down"): + raise requests.ConnectionError("down") + return make_response(404, b"nope", url=url) + + session = FakeSession(handler) + with pytest.raises(requests.ConnectionError): + service.get(session, "https://api.test/down") + service.get(session, "https://api.test/missing") + totals = _counters(service) + assert totals["requests"] == 2 + assert totals["errors"] == 1 + assert totals["http_errors"] == 1 + + def test_every_change_bumps_the_change_count(self, service): + before = service.change_count + service.get(FakeSession(), "https://api.test/x") + assert service.change_count > before + + def test_post_is_counted_and_never_merged(self, service): + class PostSession(FakeSession): + def post(self, url, **kwargs): + self.calls.append((url, kwargs)) + return make_response(201, b"{}", url=url) + + session = PostSession() + with plugin_scope("poster"): + response = service.post(session, "https://api.test/x", json={"a": 1}) + assert response.status_code == 201 + assert session.calls == [("https://api.test/x", {"json": {"a": 1}})] + assert _counters(service, plugin="poster")["requests"] == 1 + + +class TestCallerIdentity: + + def test_scope_wins_and_nests(self): + assert current_plugin_id() is None + with plugin_scope("outer"): + assert current_plugin_id() == "outer" + with plugin_scope("inner"): + assert current_plugin_id() == "inner" + with plugin_scope(None): + assert current_plugin_id() == "outer" + assert current_plugin_id() is None + + def test_a_plugins_own_thread_is_named_by_its_source_directory(self, tmp_path, service): + plugin_dir = tmp_path / "my-plugin" + plugin_dir.mkdir() + (plugin_dir / "fetcher.py").write_text( + "import threading\n" + "def fetch_in_thread(service, session, url):\n" + " t = threading.Thread(target=lambda: service.get(session, url))\n" + " t.start()\n" + " t.join(5)\n", + encoding="utf-8") + spec = importlib.util.spec_from_file_location("_fs_test_fetcher", plugin_dir / "fetcher.py") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + register_plugin_directory("my-plugin", plugin_dir) + try: + module.fetch_in_thread(service, FakeSession(), "https://api.test/x") + finally: + unregister_plugin_directory("my-plugin") + assert _counters(service, plugin="my-plugin")["requests"] == 1 + assert "core" not in service.snapshot()["plugins"] + + def test_the_executor_scopes_a_plugin_operation(self): + from src.plugin_system.plugin_executor import PluginExecutor + seen = PluginExecutor().execute_with_timeout(current_plugin_id, plugin_id="clock") + assert seen == "clock" + + def test_background_fetches_count_against_the_submitter(self, global_service): + from unittest.mock import MagicMock + from src.background_data_service import BackgroundDataService + + cache = MagicMock() + cache.get.return_value = None + bds = BackgroundDataService(cache, max_workers=1, request_timeout=5) + bds.session = FakeSession(lambda url, kw: make_response(body=b'{"events": []}', url=url)) + try: + with plugin_scope("football-scoreboard"): + request_id = bds.submit_fetch_request( + "nfl", 2026, "https://site.api.espn.com/apis/site/v2/sports/football/nfl/scoreboard", + cache_key="fs_test_nfl", params={"dates": "2026"}) + deadline = time.monotonic() + 5 + while not bds.is_request_complete(request_id) and time.monotonic() < deadline: + time.sleep(0.01) + assert bds.get_result(request_id).success + finally: + bds.shutdown(wait=True) + assert _counters(global_service, plugin="football-scoreboard")["requests"] == 1 + + def test_espn_chunks_on_worker_threads_count_against_the_caller(self, global_service): + from src.common.espn_dates import espn_date_chunks, fetch_espn_date_chunks, parse_espn_date_range + + session = FakeSession(lambda url, kw: make_response(body=b'{"events": []}', url=url)) + dates = "20260801-20261015" + with plugin_scope("baseball-scoreboard"): + fetch_espn_date_chunks(session, "https://site.api.espn.com/s/scoreboard", + params={"dates": dates}) + chunks = len(espn_date_chunks(*parse_espn_date_range(dates))) + assert chunks > 1 + assert len(session.calls) == chunks + assert _counters(global_service, plugin="baseball-scoreboard")["requests"] == chunks + assert "core" not in global_service.snapshot()["plugins"] + + def test_api_helper_goes_through_the_service(self, global_service): + from src.common.api_helper import APIHelper + + helper = APIHelper() + helper.set_rate_limit(0) + helper.session = FakeSession(lambda url, kw: make_response(body=b'{"a": 1}', url=url)) + with plugin_scope("nfl-draft"): + assert helper.get("https://api.test/x") == {"a": 1} + assert _counters(global_service, plugin="nfl-draft")["requests"] == 1 + + def test_odds_go_through_the_service(self, global_service): + from unittest.mock import MagicMock + from src.base_odds_manager import BaseOddsManager + + cache = MagicMock() + cache.get_with_auto_strategy.return_value = None + manager = BaseOddsManager(cache) + manager.session = FakeSession( + lambda url, kw: make_response(body=b'{"count": 0, "items": []}', url=url)) + with plugin_scope("odds-ticker"): + assert manager.get_odds("football", "nfl", "401") is None + assert manager.session.calls[0][1] == {"timeout": manager.request_timeout} + assert _counters(global_service, plugin="odds-ticker")["requests"] == 1 + + +# --- pooling ------------------------------------------------------------------------------------------- + +class TestConnectionPool: + + def test_core_sessions_with_one_policy_share_one_adapter(self, global_service): + from unittest.mock import MagicMock + from src.background_data_service import BackgroundDataService + from src.base_odds_manager import BaseOddsManager + + odds_a = BaseOddsManager(MagicMock()).session.get_adapter("https://x.test") + odds_b = BaseOddsManager(MagicMock()).session.get_adapter("https://x.test") + bds = BackgroundDataService(MagicMock(), max_workers=1) + try: + assert odds_a is odds_b is bds.session.get_adapter("https://x.test") + assert odds_a.max_retries.total == 0 + finally: + bds.shutdown(wait=False) + + def test_a_different_retry_policy_gets_its_own_adapter(self, global_service): + from src.common.api_helper import APIHelper + helper_adapter = APIHelper().session.get_adapter("https://x.test") + assert helper_adapter is APIHelper().session.get_adapter("https://x.test") + assert helper_adapter is not global_service.shared_adapter(0) + assert helper_adapter.max_retries.total == 3 + assert helper_adapter.max_retries.status_forcelist == [429, 500, 502, 503, 504] + assert APIHelper(max_retries=1).session.get_adapter("https://x.test") is not helper_adapter + + def test_the_pooled_session_keeps_no_cookies(self, service): + import http.client + import io + from types import SimpleNamespace + from requests.cookies import extract_cookies_to_jar + + def offer_cookie(session): + msg = http.client.parse_headers(io.BytesIO(b"Set-Cookie: sid=1; Path=/" + b"\r\n" * 2)) + raw = SimpleNamespace(_original_response=SimpleNamespace(msg=msg)) + request = requests.Request("GET", "https://api.test/").prepare() + extract_cookies_to_jar(session.cookies, request, raw) + return len(session.cookies) + + assert offer_cookie(requests.Session()) == 1 # what a private Session does + assert offer_cookie(service.session_for("https://api.test/")) == 0 + + +# --- configuration ---------------------------------------------------------------------------------- + +class TestConfigure: + + def test_reapplying_the_same_section_keeps_the_validator_store(self, clock): + config = {"rate_limits": {}} + svc = FetchService(config, clock=clock.now, sleep=clock.sleep) + svc.get(FakeSession(Versioned()), "https://api.test/x") + svc.configure(dict(config)) + assert svc.snapshot()["validators"]["entries"] == 1 + svc.configure({"rate_limits": {"api.test": {"per_second": 1}}}) + assert svc.snapshot()["validators"]["entries"] == 0 + + @pytest.mark.parametrize("bad", [ + "nonsense", + {"rate_limits": "nonsense"}, + {"rate_limits": {"api.test": "fast"}}, + {"rate_limits": {"api.test": {"per_second": -1}}}, + {"max_wait_seconds": "long"}, + ]) + def test_bad_values_fall_back_without_raising(self, clock, bad): + svc = FetchService(bad, clock=clock.now, sleep=clock.sleep) + assert svc.describe_config()["max_wait_seconds"] == 2.0 + svc.get(FakeSession(), "https://api.test/x") + + def test_the_template_section_is_what_the_code_defaults_to(self): + import os + root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + with open(os.path.join(root, "config", "config.template.json"), encoding="utf-8") as fh: + template = json.load(fh)["fetch_service"] + for key, value in template.items(): + assert fs.DEFAULT_CONFIG[key] == value + + +# --- publishing and reading ----------------------------------------------------------------------------- + +class SharedCache: + def __init__(self): + self.entries = {} + self.writes = 0 + + def get(self, key, max_age=None, memory_ttl=None): + return self.entries.get(key) + + def set(self, key, value, *args, **kwargs): + self.writes += 1 + self.entries[key] = json.loads(json.dumps(value)) + + +class TestPublisher: + + def _publisher(self, service, clock, cache): + return FetchStatsPublisher(cache, service, clock=clock.now, wall_clock=lambda: 5000.0) + + def test_on_change_at_most_once_a_minute(self, service, clock): + cache = SharedCache() + publisher = self._publisher(service, clock, cache) + assert publisher.tick() is True # first publish + assert publisher.tick() is False # nothing changed + service.get(FakeSession(), "https://api.test/x") + clock.advance(30) + assert publisher.tick() is False # changed, but too soon + clock.advance(30) + assert publisher.tick() is True + assert cache.writes == 2 + snap = cache.entries[fs.FETCH_STATS_KEY] + assert snap["running"] is True + assert snap["totals"]["requests"] == 1 + + def test_heartbeat_when_nothing_changes(self, service, clock): + cache = SharedCache() + publisher = self._publisher(service, clock, cache) + publisher.tick() + clock.advance(fs.REFRESH_INTERVAL - 1) + assert publisher.tick() is False + clock.advance(1) + assert publisher.tick() is True + + def test_stop_publishes_stopped(self, service, clock): + cache = SharedCache() + publisher = self._publisher(service, clock, cache) + publisher.stop() + assert cache.entries[fs.FETCH_STATS_KEY]["running"] is False + + def test_a_failing_cache_never_raises(self, service, clock): + class Broken(SharedCache): + def set(self, *a, **k): + raise OSError("disk full") + + assert self._publisher(service, clock, Broken()).tick() is False + + def test_reader_statuses(self, service, clock): + cache = SharedCache() + assert read_fetch_stats(cache)["status"] == "unknown" + assert read_fetch_stats(None)["status"] == "unknown" + publisher = self._publisher(service, clock, cache) + publisher.tick() + assert read_fetch_stats(cache, now=5010.0)["status"] == "live" + assert read_fetch_stats(cache, now=5000.0 + fs.STALE_AFTER + 1)["status"] == "stale" + publisher.stop() + view = read_fetch_stats(cache, now=5010.0) + assert view["status"] == "stopped" + assert view["data"]["totals"]["requests"] == 0 + + +def test_the_web_route_returns_the_published_counters(clock): + from test._api_v3_test_helpers import build_app + from web_interface.blueprints import api_v3 as module + + svc = FetchService({"rate_limits": {}}, clock=clock.now, sleep=clock.sleep) + with plugin_scope("weather"): + svc.get(FakeSession(), "https://api.test/x") + cache = SharedCache() + FetchStatsPublisher(cache, svc, wall_clock=time.time).tick() + + original = getattr(module.api_v3, "cache_manager", None) + module.api_v3.cache_manager = cache + try: + body = build_app(module.api_v3).test_client().get("/api/v3/plugins/fetch-stats").get_json() + finally: + module.api_v3.cache_manager = original + assert body["status"] == "success" + assert body["data"]["status"] == "live" + assert body["data"]["data"]["plugins"]["weather"]["requests"] == 1 diff --git a/web_interface/blueprints/api_v3/plugin_health.py b/web_interface/blueprints/api_v3/plugin_health.py index f38e8795..415f0f5d 100644 --- a/web_interface/blueprints/api_v3/plugin_health.py +++ b/web_interface/blueprints/api_v3/plugin_health.py @@ -1,4 +1,4 @@ -"""Plugin health, resource metrics and resource limits. +"""Plugin health, resource metrics, fetch statistics and resource limits. The display process records health and metrics to the shared on-disk cache; these routes read (and reset) that published state through a tracker and a @@ -193,6 +193,24 @@ def reset_plugin_metrics(plugin_id): }) +@api_v3.route('/plugins/fetch-stats', methods=['GET']) +def get_fetch_stats(): + """Network requests per plugin and per host, as the display counts them. + + Read-only. The display's fetch service (src/common/fetch_service.py) + publishes its counters to the shared cache at most once a minute when + they change; this returns that snapshot judged for staleness: + ``data.status`` is ``live``, ``stale``, ``stopped`` or ``unknown``, and + ``data.data`` the snapshot (None when unknown). Counters are cumulative + since the display started. + """ + from src.common.fetch_service import read_fetch_stats + return jsonify({ + 'status': 'success', + 'data': read_fetch_stats(getattr(api_v3, 'cache_manager', None)), + }) + + @api_v3.route('/plugins/limits/', methods=['GET', 'POST']) def manage_plugin_limits(plugin_id): """Get or set resource limits for a plugin.