added bounded close session for curl_cffi/past_prices to pricing model

This commit is contained in:
TPV committed 2026-07-28 09:15:38 -07:00
1 parent c29032a765
commit 8a1c1893b7
4 files changed
+191 -3

No files matched your search

+4
View File
@@ -41,6 +41,10 @@ CYCLE_TIMEOUT_SECONDS = 600
# None placeholders until popped), so a generous ceiling costs a low-concurrency # None placeholders until popped), so a generous ceiling costs a low-concurrency
# monitor nothing. Monitors fanning out past this pass max_clients explicitly. # monitor nothing. Monitors fanning out past this pass max_clients explicitly.
MAX_CURL_HANDLES = 64 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_IDENTIFIERS = ["chrome", "chrome_android", "safari", "safari_ios"]
CLIENT_TIMEOUT = aiohttp.ClientTimeout( CLIENT_TIMEOUT = aiohttp.ClientTimeout(
total=TIMEOUT_SECONDS, sock_connect=30, sock_read=30 total=TIMEOUT_SECONDS, sock_connect=30, sock_read=30
+2
View File
@@ -54,12 +54,14 @@ class PricingInfo:
current_price: Current listed price. current_price: Current listed price.
price_before_drop: Stored price prior to the last detected drop. price_before_drop: Stored price prior to the last detected drop.
edu_price: Education-store price when offered. edu_price: Education-store price when offered.
past_prices: Prior price history as price/date entries, oldest first.
""" """
original_price: float original_price: float
current_price: float current_price: float
price_before_drop: float | None = None price_before_drop: float | None = None
edu_price: float | None = None edu_price: float | None = None
past_prices: list[dict] | None = None
@dataclass @dataclass
+50 -3
View File
@@ -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. return values so monitor call sites change as little as possible.
""" """
import asyncio
import logging
import random import random
from aioproxies import AioProxies from aioproxies import AioProxies
@@ -15,9 +17,53 @@ from monitor_utils.constants import (
CLIENT_IDENTIFIERS, CLIENT_IDENTIFIERS,
MAX_CURL_HANDLES, MAX_CURL_HANDLES,
RETRY_LIMIT, RETRY_LIMIT,
SESSION_CLOSE_TIMEOUT_SECONDS,
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( def get_default_session(
proxies: AioProxies, proxies: AioProxies,
@@ -54,11 +100,12 @@ def get_default_session(
requests) so the wait is never reachable. requests) so the wait is never reachable.
Returns: Returns:
A TLSSession over a curl_cffi backend with a random impersonation, with A BoundedCloseSession (a TLSSession whose close is bounded) over a
the proxy pool attached as proxy_pool. curl_cffi backend with a random impersonation, with the proxy pool
attached as proxy_pool.
""" """
impersonate = random.choice(CLIENT_IDENTIFIERS) impersonate = random.choice(CLIENT_IDENTIFIERS)
session = TLSSession( session = BoundedCloseSession(
backend=CurlCffi(impersonate=impersonate), backend=CurlCffi(impersonate=impersonate),
headers=headers, headers=headers,
timeout=timeout, timeout=timeout,
+135
View File
@@ -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())