added no msrp prirority so they don't starve in queue

build/lib/monitor_utils/sessions.py

@@ -0,0 +1,173 @@
+"""Shared HTTP session helpers for all monitors.
+
+A single aioweb_tls TLSSession (curl_cffi backend, random client fingerprint per
+session) replaces the old aiohttp/curl_cffi split. send_request wraps aioweb's
+request_with_retries, adding per-request proxy rotation and old-style JSON/text
+return values so monitor call sites change as little as possible.
+"""
+
+import asyncio
+import logging
+import random
+
+from aioproxies import AioProxies
+from aioweb import Response
+from aioweb_tls import CurlCffi, TLSSession
+from monitor_utils.constants import (
+    CLIENT_IDENTIFIERS,
+    MAX_CURL_HANDLES,
+    RETRY_LIMIT,
+    SESSION_CLOSE_TIMEOUT_SECONDS,
+    TIMEOUT_SECONDS,
+)
+
+LOGGER = logging.getLogger(__name__)
+
+
+class BoundedCloseSession(TLSSession):
+    """A TLSSession whose close cannot outlive SESSION_CLOSE_TIMEOUT_SECONDS.
+
+    curl_cffi's close awaits the multi handle with no timeout of its own, so a
+    wedged transfer parks whatever is closing the session. A caller-side budget
+    cannot cover this reliably: asyncio.timeout cancels its task once, so a
+    cycle budget that has already fired leaves the close running as a fresh,
+    uncancelled await, and the process parks in the `async with` exit that was
+    meant to reclaim it.
+
+    Bounding it here keeps that guarantee attached to the session itself, so
+    every `async with get_default_session(...)` gets it without the call site
+    having to arrange anything. Abandoning a session leaks its handle until the
+    process restarts, which is strictly better than a wedged monitor.
+
+    Known inconsistency, deliberate: an abandoned session reports itself closed
+    while its curl handle is still open. TLSSession.close sets _closed before
+    awaiting the backend, so _is_closed() returns True on this path, and since
+    aioweb's __del__ only warns when _is_closed() is False, its leak warning is
+    suppressed for exactly the sessions that leaked. The "session close exceeded
+    budget" error logged below is therefore the only signal that it happened —
+    alert on it rather than on the library's warning.
+
+    Correcting the flag would mean overriding _is_closed too, and aioweb_tls is
+    a third-party library: this class deliberately overrides one documented
+    lifecycle method and nothing private, so an upstream release has one narrow
+    way to break it. A stale flag on a session discarded the same instant is the
+    cheaper side of that trade.
+    """
+
+    async def close(self) -> None:
+        try:
+            async with asyncio.timeout(SESSION_CLOSE_TIMEOUT_SECONDS):
+                await super().close()
+        except TimeoutError:
+            LOGGER.error(
+                "session close exceeded budget, abandoning session",
+                extra={"timeout": SESSION_CLOSE_TIMEOUT_SECONDS},
+            )
+
+
+def get_default_session(
+    proxies: AioProxies,
+    headers: dict | None = None,
+    timeout: int = TIMEOUT_SECONDS,
+    max_clients: int = MAX_CURL_HANDLES,
+) -> TLSSession:
+    """Return a TLSSession impersonating a random client fingerprint.
+
+    Fingerprint rotation lives here so every monitor rotates the same way; the
+    caller owns the session lifetime via ``async with``.
+
+    Cookies are not set here: aioweb has no constructor cookie path, and
+    monitors usually obtain them at runtime, so call session.set_cookie(name,
+    value, domain=...) where they are fetched (e.g. a pre_task_func) instead.
+
+    Args:
+        proxies: Pool attached to the session as proxy_pool, which send_request
+            rotates through per request. Required so a session can never send
+            un-proxied.
+        headers: Default headers merged into every request by aioweb. Use for
+            monitor-wide static headers; per-request headers still go through
+            send_request. The curl_cffi impersonation already supplies a
+            matching User-Agent, so one is not defaulted here.
+        timeout: Total request timeout in seconds, defaulting to TIMEOUT_SECONDS.
+            aioweb's own default is only 10s, so this restores the wider budget
+            the old sessions used and slow retailers depend on.
+        max_clients: Size of the curl handle pool, defaulting to
+            MAX_CURL_HANDLES. This is a hard concurrency ceiling: a request
+            finding the pool empty waits in pop_curl's untimed pool.get(), and
+            `timeout` does not cover that wait, only the transfer after
+            checkout. Keep it at or above the monitor's real fan-out (the sum
+            of its pagination semaphores plus any unsemaphored per-cycle
+            requests) so the wait is never reachable.
+
+    Returns:
+        A BoundedCloseSession (a TLSSession whose close is bounded) over a
+        curl_cffi backend with a random impersonation, with the proxy pool
+        attached as proxy_pool.
+    """
+    impersonate = random.choice(CLIENT_IDENTIFIERS)
+    session = BoundedCloseSession(
+        backend=CurlCffi(impersonate=impersonate),
+        headers=headers,
+        timeout=timeout,
+        max_clients=max_clients,
+    )
+    session.proxy_pool = proxies
+    return session
+
+
+async def send_request(
+    session: TLSSession,
+    url: str,
+    method: str,
+    *,
+    params: dict | None = None,
+    json_data: dict | None = None,
+    data: dict | None = None,
+    headers: dict | None = None,
+) -> Response:
+    """Send a request through a TLSSession with retries and proxy rotation.
+
+    Wraps aioweb's request_with_retries, which retries transport errors,
+    timeouts, and retryable statuses (429/5xx) up to RETRY_LIMIT attempts with
+    exponential backoff. The next proxy is drawn from the pool per call and
+    applied to both schemes. A non-2xx response or a fully failed request raises
+    AiowebError, an expected task error, so a failed fetch surfaces the same way
+    the old HTTPError path did. On success the aioweb Response is returned for
+    the caller to read via response.json(), response.text(), etc.
+
+    Args:
+        session: The TLSSession to send through.
+        url: Request URL.
+        method: HTTP method.
+        params: Query parameters.
+        json_data: JSON body. A dict here is sent as JSON with the
+            application/json content-type set by aioweb.
+        data: Raw request body, passed through unchanged (e.g. a url-encoded
+            form string). The caller owns the matching content-type via headers.
+            A dict is JSON-encoded by aioweb, so use json_data for JSON bodies.
+        headers: Per-request headers merged over the session defaults.
+
+    Returns:
+        The aioweb Response, guaranteed 2xx (non-2xx raises). Read the body with
+        response.json() (dict/list, or None if not valid JSON) or
+        response.text().
+
+    Raises:
+        AiowebError: On a non-2xx response or after all retries fail.
+    """
+
+    proxy_url = session.proxy_pool.next().url()
+    proxy = {"http": proxy_url, "https": proxy_url}
+
+    body = json_data if json_data is not None else data
+    response = await session.request_with_retries(
+        method.upper(),
+        url,
+        params=params,
+        data=body,
+        headers=headers,
+        proxies=proxy,
+        attempts=RETRY_LIMIT,
+    )
+    response.raise_for_status()
+    return response