diff --git a/src/monitor_utils/constants.py b/src/monitor_utils/constants.py index f1f801f..c48f0b9 100644 --- a/src/monitor_utils/constants.py +++ b/src/monitor_utils/constants.py @@ -41,6 +41,10 @@ CYCLE_TIMEOUT_SECONDS = 600 # None placeholders until popped), so a generous ceiling costs a low-concurrency # monitor nothing. Monitors fanning out past this pass max_clients explicitly. MAX_CURL_HANDLES = 64 +# Budget for closing a session. curl_cffi's close awaits the multi handle with +# no timeout of its own, so a wedged transfer would otherwise park whatever is +# closing the session, including an `async with` exit. +SESSION_CLOSE_TIMEOUT_SECONDS = 10 CLIENT_IDENTIFIERS = ["chrome", "chrome_android", "safari", "safari_ios"] CLIENT_TIMEOUT = aiohttp.ClientTimeout( total=TIMEOUT_SECONDS, sock_connect=30, sock_read=30 diff --git a/src/monitor_utils/models.py b/src/monitor_utils/models.py index ebd5c13..d103973 100644 --- a/src/monitor_utils/models.py +++ b/src/monitor_utils/models.py @@ -54,12 +54,14 @@ class PricingInfo: current_price: Current listed price. price_before_drop: Stored price prior to the last detected drop. edu_price: Education-store price when offered. + past_prices: Prior price history as price/date entries, oldest first. """ original_price: float current_price: float price_before_drop: float | None = None edu_price: float | None = None + past_prices: list[dict] | None = None @dataclass diff --git a/src/monitor_utils/sessions.py b/src/monitor_utils/sessions.py index 377eeee..c909372 100644 --- a/src/monitor_utils/sessions.py +++ b/src/monitor_utils/sessions.py @@ -6,6 +6,8 @@ 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 @@ -15,9 +17,53 @@ 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, @@ -54,11 +100,12 @@ def get_default_session( requests) so the wait is never reachable. Returns: - A TLSSession over a curl_cffi backend with a random impersonation, with - the proxy pool attached as proxy_pool. + 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 = TLSSession( + session = BoundedCloseSession( backend=CurlCffi(impersonate=impersonate), headers=headers, timeout=timeout, diff --git a/tests/test_session_close_budget.py b/tests/test_session_close_budget.py new file mode 100644 index 0000000..5421777 --- /dev/null +++ b/tests/test_session_close_budget.py @@ -0,0 +1,135 @@ +"""unit test: a session's close cannot park the caller indefinitely + +run from the repo root: + uv run pytest tests/test_session_close_budget.py + +offline — no network. curl_cffi's close awaits the multi handle with no timeout +of its own, so a wedged transfer would park whatever closes the session, +including an `async with` exit. These build a real session and wedge the +backend's close, asserting the bound lives on the session itself so every call +site gets it without arranging anything. +""" + +import asyncio +import logging + +from aioproxies import AioProxies + +from monitor_utils import sessions +from monitor_utils.sessions import BoundedCloseSession, get_default_session + + +def _proxies(tmp_path) -> AioProxies: + """A one-entry pool in the host:port:user:pass format monitors ship.""" + proxy_file = tmp_path / "proxies.txt" + proxy_file.write_text("proxy.example.io:8080:user:secret\n") + return AioProxies.from_file(path=str(proxy_file)) + + +async def _wedge_close(session) -> asyncio.Event: + """Replace the backend's close with one that never returns. + + Builds the backend session first so TLSSession.close actually reaches the + backend rather than short-circuiting on an unbuilt session. + """ + + session.session # force lazy construction + started = asyncio.Event() + + async def wedged(inner_session): + started.set() + await asyncio.sleep(3600) + + session.backend.close = wedged + return started + + +async def _really_close(session, original_close): + """Close for real, so the test does not leak the curl handle it allocated.""" + session.backend.close = original_close + await session.close() + + +def test_close_is_abandoned_when_it_exceeds_the_budget(monkeypatch, caplog, tmp_path): + """A close that never returns is abandoned and logged, not awaited forever.""" + + monkeypatch.setattr(sessions, "SESSION_CLOSE_TIMEOUT_SECONDS", 0.05) + proxies = _proxies(tmp_path) + + async def run(): + session = get_default_session(proxies, timeout=8) + original_close = session.backend.close + started = await _wedge_close(session) + + # wait_for is the assertion: an unbounded close never returns here + await asyncio.wait_for(session.close(), timeout=5) + + assert started.is_set() + await _really_close(session, original_close) + + with caplog.at_level(logging.ERROR, logger="monitor_utils.sessions"): + asyncio.run(run()) + + assert any( + record.message == "session close exceeded budget, abandoning session" + for record in caplog.records + ) + + +def test_async_with_exit_is_bounded(monkeypatch, caplog, tmp_path): + """The bound applies through `async with`, which is how monitors close.""" + + monkeypatch.setattr(sessions, "SESSION_CLOSE_TIMEOUT_SECONDS", 0.05) + proxies = _proxies(tmp_path) + + async def run(): + session = get_default_session(proxies, timeout=8) + original_close = session.backend.close + started = await _wedge_close(session) + + async with session: + pass + + assert started.is_set() + await _really_close(session, original_close) + + with caplog.at_level(logging.ERROR, logger="monitor_utils.sessions"): + asyncio.run(asyncio.wait_for(run(), timeout=5)) + + assert any( + record.message == "session close exceeded budget, abandoning session" + for record in caplog.records + ) + + +def test_healthy_close_completes_and_is_not_logged(caplog, tmp_path): + """The normal path closes fully and reports nothing.""" + + proxies = _proxies(tmp_path) + + async def run(): + session = get_default_session(proxies, timeout=8) + session.session # force lazy construction so close has work to do + await session.close() + return session + + with caplog.at_level(logging.ERROR, logger="monitor_utils.sessions"): + session = asyncio.run(run()) + + assert session._is_closed() + assert not caplog.records + + +def test_get_default_session_returns_a_bounded_session(tmp_path): + """The factory hands out the bounded type, not a bare TLSSession.""" + + proxies = _proxies(tmp_path) + + async def run(): + session = get_default_session(proxies, timeout=8) + try: + return isinstance(session, BoundedCloseSession) + finally: + await session.close() + + assert asyncio.run(run())