From a0b077278b0534f3006444650f3c81983fcb7d1d Mon Sep 17 00:00:00 2001 From: Daniel Date: Wed, 9 Sep 2026 13:43:09 +0200 Subject: [PATCH] perf(marketguard): stream the auction house instead of collecting it The Lowest BIN refresh fetched every page of the Hypixel auction house, collected all ~100k auctions in one list and only then indexed them. The payloads carry base64 NBT blobs, so the process peaked around 700 MiB for an index that is a few hundred kilobytes. On the 2 GiB production host the kernel OOM-killed uvicorn - PID 1 in the container - roughly every five minutes; the uptime monitor reported that as hours of downtime. Pages are now folded into the index as they arrive and dropped immediately. Peak allocation follows the pages in flight rather than the size of the house: measured flat at 8.1 MiB from 4k to 32k auctions, where collecting grew 9.5 -> 20.9 MiB over the same range. Fetching stays concurrent, but as a worker pool rather than one task per page. Scheduling every page up front bounds the concurrent requests and nothing else: completed tasks hold their pages until the consumer gets to them, and since decoding is the slow part they queue up anyway. A worker only fetches its next page once it has handed the current one over, which is what actually applies back pressure. Consumption is serialised behind a lock because all workers fold into one shared index. Drift handling is unchanged in behaviour: pages are still checked against page 0, and the client now resets the consumer before each attempt so a retry cannot inherit what the discarded attempt already folded in. build_lowestbin_index() is kept for callers that already hold every auction; it now delegates to the same incremental builder. Co-Authored-By: Claude Opus 5 --- app/marketguard_api/client.py | 129 ++++++++++++++---- app/marketguard_api/models.py | 13 ++ app/marketguard_api/service.py | 102 ++++++++++++--- tests/test_marketguard_api.py | 231 ++++++++++++++++++++++++++++++++- 4 files changed, 424 insertions(+), 51 deletions(-) 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}" + )