LEDMatrix/src/common/fetch_service.py

"""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}