diff --git a/app/marketguard_api/client.py b/app/marketguard_api/client.py index 0aeb04a..737283e 100644 --- a/app/marketguard_api/client.py +++ b/app/marketguard_api/client.py @@ -6,7 +6,7 @@ import time from collections import deque from collections.abc import Callable -from typing import Any +from typing import Any, Protocol import httpx @@ -18,7 +18,7 @@ HypixelUpstreamError, MojangUpstreamError, ) -from .models import AuctionPage, AuctionSnapshot, BazaarProductSnapshot +from .models import AuctionPage, AuctionSnapshot, AuctionSnapshotSummary, BazaarProductSnapshot logger = logging.getLogger(__name__) @@ -61,6 +61,31 @@ async def acquire(self) -> None: _HYPIXEL_KEY_RATE_LIMITER = _HypixelKeyRateLimiter() +class AuctionPageConsumer(Protocol): + """Folds auction pages as they arrive so no caller holds the whole house. + + ``reset`` is called before every snapshot attempt, because a drift retry + has to discard whatever the previous attempt already folded in. + """ + + def reset(self) -> None: ... + + async def add_page(self, auctions: list[dict[str, Any]]) -> None: ... + + +class _CollectingConsumer: + """Consumer that keeps every auction, for callers that really want them all.""" + + def __init__(self) -> None: + self.auctions: list[dict[str, Any]] = [] + + def reset(self) -> None: + self.auctions = [] + + async def add_page(self, auctions: list[dict[str, Any]]) -> None: + self.auctions.extend(auctions) + + class HypixelAuctionClient: def __init__( self, @@ -98,11 +123,20 @@ async def aclose(self) -> None: await self._client.aclose() self._client = None - async def fetch_snapshot(self) -> AuctionSnapshot: + async def stream_snapshot(self, consumer: AuctionPageConsumer) -> AuctionSnapshotSummary: + """Fetch one consistent snapshot, handing each page to ``consumer``. + + Pages are still fetched concurrently; they are just consumed as they + land instead of being collected first. Holding all ~100k auctions at + once cost roughly 700 MiB and got the process OOM-killed on a 2 GiB + host, so nothing here may keep a page alive past ``add_page``. + """ last_error: Exception | None = None for attempt in range(1, self._settings.snapshot_retries + 1): + # A retry must not inherit the pages the failed attempt folded in. + consumer.reset() try: - return await self._fetch_consistent_snapshot() + return await self._stream_consistent_snapshot(consumer) except HypixelSnapshotDriftError as exc: last_error = exc if attempt >= self._settings.snapshot_retries: @@ -114,32 +148,71 @@ async def fetch_snapshot(self) -> AuctionSnapshot: ) raise HypixelUpstreamError("Unable to obtain a consistent Hypixel auction snapshot.") from last_error - async def _fetch_consistent_snapshot(self) -> AuctionSnapshot: - first_page = await self._fetch_page(0) - auctions = list(first_page.auctions) - - if first_page.total_pages > 1: - semaphore = asyncio.Semaphore(self._settings.max_parallel_pages) + async def fetch_snapshot(self) -> AuctionSnapshot: + """Fetch a snapshot and keep every auction. Prefer ``stream_snapshot``.""" + consumer = _CollectingConsumer() + summary = await self.stream_snapshot(consumer) + return AuctionSnapshot( + total_pages=summary.total_pages, + last_updated=summary.last_updated, + auctions=consumer.auctions, + ) - async def _fetch_followup(page_number: int) -> AuctionPage: - async with semaphore: + async def _stream_consistent_snapshot(self, consumer: AuctionPageConsumer) -> AuctionSnapshotSummary: + first_page = await self._fetch_page(0) + total_pages = first_page.total_pages + last_updated = first_page.last_updated + total_auctions = len(first_page.auctions) + await consumer.add_page(first_page.auctions) + # Drop the first page before fetching the rest; the counts above are + # all that is still needed from it. + del first_page + + if total_pages > 1: + # A fixed pool of workers, each holding at most one page: a worker + # only fetches its next page once it has handed the current one + # over. Scheduling every page up front instead would bound the + # concurrent *requests* but not the pages already fetched - those + # pile up while the consumer is busy decoding, which is exactly the + # accumulation this streaming path exists to avoid. + page_numbers = iter(range(1, total_pages)) + # Consumption is serialised: the consumer folds into one shared + # index, and two folds at once would corrupt it. + consume_lock = asyncio.Lock() + + async def _worker() -> None: + nonlocal total_auctions + # Advancing a plain iterator is safe here: asyncio runs these + # workers on one thread and never preempts between the next() + # and the await that follows it. + for page_number in page_numbers: page = await self._fetch_page(page_number) - if page.last_updated != first_page.last_updated: - raise HypixelSnapshotDriftError( - f"Hypixel snapshot drift detected between page 0 and page {page_number}." - ) - return page - - pages = await asyncio.gather( - *(_fetch_followup(page_number) for page_number in range(1, first_page.total_pages)) - ) - for page in pages: - auctions.extend(page.auctions) - - return AuctionSnapshot( - total_pages=first_page.total_pages, - last_updated=first_page.last_updated, - auctions=auctions, + if page.last_updated != last_updated: + raise HypixelSnapshotDriftError( + f"Hypixel snapshot drift detected between page 0 and page {page_number}." + ) + async with consume_lock: + total_auctions += len(page.auctions) + await consumer.add_page(page.auctions) + # Drop it before fetching the next one. + del page + + worker_count = min(self._settings.max_parallel_pages, total_pages - 1) + workers = [asyncio.ensure_future(_worker()) for _ in range(worker_count)] + try: + await asyncio.gather(*workers) + finally: + # On drift (or any failure) the remaining workers would keep + # fetching pages nobody wants; stop them right away. + for worker in workers: + if not worker.done(): + worker.cancel() + await asyncio.gather(*workers, return_exceptions=True) + + return AuctionSnapshotSummary( + total_pages=total_pages, + last_updated=last_updated, + total_auctions=total_auctions, ) async def _fetch_page(self, page_number: int) -> AuctionPage: diff --git a/app/marketguard_api/models.py b/app/marketguard_api/models.py index 4f773dc..b541e49 100644 --- a/app/marketguard_api/models.py +++ b/app/marketguard_api/models.py @@ -28,6 +28,19 @@ class AuctionSnapshot: auctions: list[dict[str, Any]] +@dataclass(frozen=True, slots=True) +class AuctionSnapshotSummary: + """What is left of an auction snapshot once its pages have been consumed. + + Streaming the auction house means no caller holds the ~100k auction + payloads afterwards, so the counts have to be carried out separately. + """ + + total_pages: int + last_updated: int + total_auctions: int + + @dataclass(frozen=True, slots=True) class BazaarProductSnapshot: last_updated: int diff --git a/app/marketguard_api/service.py b/app/marketguard_api/service.py index b0208c2..e4f789f 100644 --- a/app/marketguard_api/service.py +++ b/app/marketguard_api/service.py @@ -33,8 +33,43 @@ class LowestBinIndex: total_bin_auctions: int -def build_lowestbin_index(auctions: list[dict[str, Any]]) -> LowestBinIndex: - """Decode every auction payload and index the lowest BIN price per item key. +class LowestBinIndexBuilder: + """Accumulates the Lowest BIN index one auction page at a time. + + The index itself is tiny - a few hundred kilobytes of item key to price. + What used to be expensive was the input: collecting all ~100k auction + payloads before indexing them pushed the process past 700 MiB and got it + OOM-killed on a 2 GiB host. Folding page by page keeps only the pages + currently in flight. + """ + + __slots__ = ("lowest_bins", "auctioneer_uuids", "item_names", "total_bin_auctions") + + def __init__(self) -> None: + self.reset() + + def reset(self) -> None: + self.lowest_bins: dict[str, float] = {} + self.auctioneer_uuids: dict[str, str] = {} + self.item_names: dict[str, str] = {} + self.total_bin_auctions = 0 + + def add_page(self, auctions: list[dict[str, Any]]) -> None: + fold_auctions_into_index(auctions, self) + + def build(self) -> LowestBinIndex: + return LowestBinIndex( + lowest_bins=dict(sorted(self.lowest_bins.items(), key=lambda item: item[0].lower())), + metadata=LowestBinSnapshotMetadata( + auctioneer_uuids=self.auctioneer_uuids, + item_names=self.item_names, + ), + total_bin_auctions=self.total_bin_auctions, + ) + + +def fold_auctions_into_index(auctions: list[dict[str, Any]], accumulator: LowestBinIndexBuilder) -> None: + """Decode one page of auctions and fold it into ``accumulator``. This is pure CPU work: a full Hypixel auction house is ~100k auctions, and each one costs a base64 decode, a gzip inflate and an NBT parse. Measured at @@ -42,10 +77,10 @@ def build_lowestbin_index(auctions: list[dict[str, Any]]) -> LowestBinIndex: doing so stalls every other request, including the health and readiness probes. Callers run it through ``asyncio.to_thread``. """ - lowest_bins: dict[str, float] = {} - auctioneer_uuids: dict[str, str] = {} - item_names: dict[str, str] = {} - total_bin_auctions = 0 + lowest_bins = accumulator.lowest_bins + auctioneer_uuids = accumulator.auctioneer_uuids + item_names = accumulator.item_names + page_bin_auctions = 0 for auction in auctions: if auction.get("bin") is not True: @@ -59,7 +94,7 @@ def build_lowestbin_index(auctions: list[dict[str, Any]]) -> LowestBinIndex: if auctioneer_uuid is None: continue - total_bin_auctions += 1 + page_bin_auctions += 1 for item_key in resolved_item.keys: current_lowest = lowest_bins.get(item_key) if current_lowest is None or resolved_item.unit_price < current_lowest: @@ -67,11 +102,38 @@ def build_lowestbin_index(auctions: list[dict[str, Any]]) -> LowestBinIndex: auctioneer_uuids[item_key] = auctioneer_uuid item_names[item_key] = _parse_item_name(auction.get("item_name"), item_key) - return LowestBinIndex( - lowest_bins=dict(sorted(lowest_bins.items(), key=lambda item: item[0].lower())), - metadata=LowestBinSnapshotMetadata(auctioneer_uuids=auctioneer_uuids, item_names=item_names), - total_bin_auctions=total_bin_auctions, - ) + accumulator.total_bin_auctions += page_bin_auctions + + +def build_lowestbin_index(auctions: list[dict[str, Any]]) -> LowestBinIndex: + """Index a complete auction list in one call. + + Kept for callers that already hold every auction; the refresh path streams + instead, via :class:`LowestBinIndexBuilder`. + """ + builder = LowestBinIndexBuilder() + builder.add_page(auctions) + return builder.build() + + +class _IndexingPageConsumer: + """Feeds streamed auction pages into a :class:`LowestBinIndexBuilder`.""" + + __slots__ = ("_builder",) + + def __init__(self) -> None: + self._builder = LowestBinIndexBuilder() + + def reset(self) -> None: + self._builder.reset() + + async def add_page(self, auctions: list[dict[str, Any]]) -> None: + # Decoding ~100k NBT item payloads is seconds of CPU; keep it off the + # event loop so health, readiness and every other route stay responsive. + await asyncio.to_thread(self._builder.add_page, auctions) + + def build(self) -> LowestBinIndex: + return self._builder.build() class LowestBinService: @@ -204,16 +266,18 @@ async def _load_snapshot(self) -> tuple[LowestBinSnapshot, LowestBinSnapshotMeta return snapshot, metadata async def _refresh_snapshot(self) -> tuple[LowestBinSnapshot, LowestBinSnapshotMetadata]: - auction_snapshot = await self._client.fetch_snapshot() - # Decoding ~100k NBT item payloads is seconds of CPU; keep it off the - # event loop so health, readiness and every other route stay responsive. - index = await asyncio.to_thread(build_lowestbin_index, auction_snapshot.auctions) + # Stream the auction house instead of collecting it: the payloads only + # matter while they are being decoded, and holding all of them at once + # is what made the host OOM-kill this process. + consumer = _IndexingPageConsumer() + summary = await self._client.stream_snapshot(consumer) + index = consumer.build() snapshot = LowestBinSnapshot( generated_at=_clock_datetime(self._clock()), - snapshot_last_updated=auction_snapshot.last_updated, - total_pages=auction_snapshot.total_pages, - total_auctions=len(auction_snapshot.auctions), + snapshot_last_updated=summary.last_updated, + total_pages=summary.total_pages, + total_auctions=summary.total_auctions, total_bin_auctions=index.total_bin_auctions, items=index.lowest_bins, is_stale=False, diff --git a/tests/test_marketguard_api.py b/tests/test_marketguard_api.py index c1f6462..debd155 100644 --- a/tests/test_marketguard_api.py +++ b/tests/test_marketguard_api.py @@ -2,10 +2,12 @@ import asyncio import base64 +import gc import gzip import hashlib import json import struct +import weakref from collections.abc import Mapping from dataclasses import replace from datetime import datetime, timedelta, timezone @@ -27,6 +29,7 @@ _HypixelKeyRateLimiter, ) from app.marketguard_api.exceptions import HypixelRateLimitError +from app.marketguard_api.models import AuctionPage from app.marketguard_api.config import MarketGuardSettings from app.marketguard_api.item_keys import resolve_auction_item from app.marketguard_api.main import create_marketguard_app @@ -2935,16 +2938,16 @@ def test_lowestbin_refresh_decodes_auctions_off_the_event_loop(monkeypatch) -> N observed: dict[str, Any] = {"decode_thread_id": None, "ticks_during_decode": 0} decode_started = threading.Event() decode_may_finish = threading.Event() - real_build_index = service_module.build_lowestbin_index + real_fold = service_module.fold_auctions_into_index - def _instrumented_build_index(auctions): + def _instrumented_fold(auctions, accumulator): observed["decode_thread_id"] = threading.get_ident() decode_started.set() # Hold the decode open so the test can prove the loop still runs. decode_may_finish.wait(timeout=5) - return real_build_index(auctions) + return real_fold(auctions, accumulator) - monkeypatch.setattr(service_module, "build_lowestbin_index", _instrumented_build_index) + monkeypatch.setattr(service_module, "fold_auctions_into_index", _instrumented_fold) async def _handler(_request: httpx.Request) -> httpx.Response: return httpx.Response( @@ -3339,3 +3342,223 @@ async def _bazaar_handler(_request: httpx.Request) -> httpx.Response: async def _unexpected_hypixel_handler(request: httpx.Request) -> httpx.Response: raise AssertionError(f"Unexpected Hypixel request: {request.url}") + + +# --- Streaming auction snapshots ------------------------------------------------- +# +# Regression cover for the OOM incident: the refresh used to collect every page +# of the auction house before indexing it. A full house is ~100k auctions whose +# payloads carry base64 NBT blobs, which put the process around 700 MiB. On a +# 2 GiB host the kernel OOM-killed uvicorn - PID 1 in the container - roughly +# every five minutes, which the uptime monitor reported as hours of downtime. + + +class _TrackedAuctions(list): + """A weakref-able auction list, so a test can prove a page was released.""" + + +class _CountingPageConsumer: + """Consumes pages without keeping them, and counts what it saw.""" + + def __init__(self) -> None: + self.pages = 0 + self.auctions = 0 + self.resets = 0 + + def reset(self) -> None: + self.pages = 0 + self.auctions = 0 + self.resets += 1 + + async def add_page(self, auctions) -> None: + self.pages += 1 + self.auctions += len(auctions) + + +def test_auction_snapshot_stream_releases_every_page() -> None: + """No auction page may outlive the streaming fetch that produced it. + + This is the whole point of streaming: holding all pages at once is what + made the host OOM-kill the API. Weak references to the page payloads prove + the client kept none of them once the snapshot finished. + """ + last_updated = _epoch_millis(2025, 3, 1, 12, 0) + total_pages = 6 + page_refs: list[weakref.ReferenceType] = [] + + async def _fetch_page(page_number: int) -> AuctionPage: + auctions = _TrackedAuctions([_auction(f"ITEM_{page_number}", 1_000 + page_number)]) + page_refs.append(weakref.ref(auctions)) + return AuctionPage( + page_number=page_number, + total_pages=total_pages, + last_updated=last_updated, + auctions=auctions, + ) + + client = HypixelAuctionClient(_marketguard_settings()) + client._fetch_page = _fetch_page # type: ignore[method-assign] + consumer = _CountingPageConsumer() + + summary = asyncio.run(client.stream_snapshot(consumer)) + + assert summary.total_pages == total_pages + assert summary.total_auctions == total_pages + assert consumer.pages == total_pages + assert len(page_refs) == total_pages + + gc.collect() + still_alive = [index for index, ref in enumerate(page_refs) if ref() is not None] + assert still_alive == [], f"auction pages {still_alive} were still referenced after streaming" + + +def test_lowestbin_refresh_indexes_every_page() -> None: + """Folding page by page must index the same auctions collecting them did.""" + last_updated = _epoch_millis(2025, 3, 1, 12, 0) + + async def _handler(request: httpx.Request) -> httpx.Response: + page = int(request.url.params.get("page", 0)) + return httpx.Response( + 200, + json={ + "success": True, + "totalPages": 3, + "lastUpdated": last_updated, + "auctions": [ + _auction(f"ITEM_{page}", 1_000 * (page + 1), item_name=f"Item {page}"), + ], + }, + ) + + settings = _marketguard_settings() + service = _marketguard_service(settings, _handler) + + async def _exercise(): + snapshot = await service.refresh() + await service.aclose() + return snapshot + + snapshot = asyncio.run(_exercise()) + + assert snapshot.total_pages == 3 + assert snapshot.total_auctions == 3 + assert snapshot.total_bin_auctions == 3 + assert snapshot.items["ITEM_0"] == 1_000.0 + assert snapshot.items["ITEM_1"] == 2_000.0 + assert snapshot.items["ITEM_2"] == 3_000.0 + + +def test_auction_snapshot_drift_discards_the_partial_index() -> None: + """A drift retry must start from an empty index, not from what it folded. + + Streaming means pages are indexed before the snapshot is known to be + consistent. Without the reset, the retry would count the discarded + attempt's auctions a second time. + """ + last_updated = _epoch_millis(2025, 3, 1, 12, 0) + state = {"page0_requests": 0} + + async def _handler(request: httpx.Request) -> httpx.Response: + page = int(request.url.params.get("page", 0)) + if page == 0: + state["page0_requests"] += 1 + # On the first attempt page 1 reports a newer snapshot, which is drift. + drifted = state["page0_requests"] == 1 and page == 1 + return httpx.Response( + 200, + json={ + "success": True, + "totalPages": 2, + "lastUpdated": last_updated + (1 if drifted else 0), + "auctions": [_auction(f"ITEM_{page}", 1_000 * (page + 1))], + }, + ) + + settings = _marketguard_settings() + service = _marketguard_service(settings, _handler) + + async def _exercise(): + snapshot = await service.refresh() + await service.aclose() + return snapshot + + snapshot = asyncio.run(_exercise()) + + assert state["page0_requests"] == 2, "the drifted attempt should have been retried" + assert snapshot.total_auctions == 2, "the discarded attempt leaked into the retry" + assert snapshot.total_bin_auctions == 2 + assert snapshot.snapshot_last_updated == last_updated + + +def test_stream_snapshot_resets_the_consumer_before_each_attempt() -> None: + """The client, not the caller, is responsible for clearing a failed attempt.""" + last_updated = _epoch_millis(2025, 3, 1, 12, 0) + attempts = {"count": 0} + + async def _fetch_page(page_number: int) -> AuctionPage: + if page_number == 0: + attempts["count"] += 1 + drifted = attempts["count"] == 1 and page_number == 1 + return AuctionPage( + page_number=page_number, + total_pages=2, + last_updated=last_updated + (1 if drifted else 0), + auctions=[_auction(f"ITEM_{page_number}", 1_000)], + ) + + client = HypixelAuctionClient(_marketguard_settings()) + client._fetch_page = _fetch_page # type: ignore[method-assign] + consumer = _CountingPageConsumer() + + summary = asyncio.run(client.stream_snapshot(consumer)) + + assert attempts["count"] == 2 + # One reset per attempt, including the one that drifted. + assert consumer.resets == 2 + assert consumer.auctions == 2 + assert summary.total_auctions == 2 + + +def test_stream_snapshot_keeps_only_the_workers_pages_in_flight() -> None: + """Peak memory must follow the worker count, not the size of the house. + + Scheduling every page up front bounds the concurrent requests but not the + pages already fetched: those queue up while the consumer decodes, which + reproduces the accumulation streaming is meant to remove. The pool must + therefore hold at most one page per worker. + """ + settings = _marketguard_settings() + total_pages = 40 + live = {"now": 0, "peak": 0} + + async def _fetch_page(page_number: int) -> AuctionPage: + # Yield the way a real HTTP round trip does, so workers interleave. + await asyncio.sleep(0) + live["now"] += 1 + live["peak"] = max(live["peak"], live["now"]) + return AuctionPage( + page_number=page_number, + total_pages=total_pages, + last_updated=_epoch_millis(2025, 3, 1, 12, 0), + auctions=[_auction(f"ITEM_{page_number}", 1_000)], + ) + + class _ReleasingConsumer: + def reset(self) -> None: + live["now"] = 0 + + async def add_page(self, auctions) -> None: + # Stand in for the decode, which is where pages used to pile up. + await asyncio.sleep(0) + live["now"] -= 1 + + client = HypixelAuctionClient(settings) + client._fetch_page = _fetch_page # type: ignore[method-assign] + + summary = asyncio.run(client.stream_snapshot(_ReleasingConsumer())) + + assert summary.total_pages == total_pages + assert live["peak"] <= settings.max_parallel_pages, ( + f"{live['peak']} pages were in flight at once, " + f"max_parallel_pages is {settings.max_parallel_pages}" + )