Stream the auction house instead of collecting it

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:

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

app/marketguard_api/service.py

@@ -33,19 +33,54 @@ 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
     5-10 seconds for a full house, so it must never run on the event loop -
     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,19 +94,46 @@ 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:
                 lowest_bins[item_key] = resolved_item.unit_price
                 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,