diff --git a/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md b/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md index 3668a9bc..7c58e66d 100644 --- a/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md +++ b/DATA_LAYER_UNIFIED_IMPLEMENTATION_PLAN.md @@ -76909,3 +76909,87 @@ locally; release image NOT rebuilt (the coordinator rebuilds rust together with until the rollback period ends; scoped cleanup) ACCEPTED: section 5c. - Status: implemented and tested locally (docs, tools); the window is NOT ready for go until the IDR-129 fix is merged, the python image rebuilt and re-pinned, and the final rehearsal passes (next entries). +### IDR-129 root cause and fix: Query RSS kept after warmup bursts - 2026-10-19 (Claude) + +- Scope: coordinator task (owner-approved fixes 2026-10-19), branch `fix/idr129-query-rss` from dev cd272f8, for + DL v2.2.5. Production read-only (docker logs/inspect/stats, one read-only key copy); host lock + `claude-idr129-query-rss` from 14:15:25Z. +- Production evidence (docker logs, qdl.stall lines): `query_kn_1` RSS stepped up per burst from 270 MiB + (10-18 05Z) to 2045 MiB (10-19 12:02Z, restart); after the restarts 190 MiB -> 871 MiB peak at 12:38Z (IDR-128 + arm B stage-100) -> 558/650 MiB at 14:12Z (docker stats). GC telemetry over 32 h: collections `[16, 0, 1]` + (16 generation-0, 0 generation-1, 1 start-up full). The bursts are 2,500/5,000/10,000-row alpha warmups + (stage-100 cold role) and 1,000-row TS BAR warmups on the single-active large-warmup lane. +- Harness `scripts/idr129_query_rss_isolated.py` (new): one Query replica of the production image 3e9e4d3a with the + production container spec except private market-cache and quota Redis, `--cpus 1.0 --memory 2g`, isolated + `--internal` network, every container `--rm`. The private cache is a read-only copy (DUMP/HMGET/HGETALL) of the + 64 burst products' keys (10,200 rows each, 218 MB) plus a 15 s tail re-copy so the 1m freshness holds. Driver = + SDK 2.4 shape (redirect 1, queued retry on REPLICA_LANE_BUSY, cold-read retry, 10 s client timeout), 64 + concurrent warmups per burst: TS 10 x 1,000 rows, alpha 54 x 2,500/5,000/10,000. A sitecustomize probe reports + RSS, mallinfo2, gc generations, row cache, and runs gc.collect (with DEBUG_SAVEALL summary) and malloc_trim(0). + Evidence `~/.local/state/qdl-v2/idr129-query-rss-20261019{,b,c}/` (arm dirs: run.json, driver.out, query.log, + work/result.json). +- Measured, production image (arm A-stage, 12 bursts): RSS 195 (start) -> 472 (burst 1) -> 683 MiB (burst 12), + malloc in use (uordblks) 155 -> 398 MiB, Python blocks flat (1.49-1.52 M), row cache at its 20,000 bound. + At idle: gc.collect freed 15,917 objects and uordblks 398 -> 90 MiB (fordblks 218 -> 526); malloc_trim(0) then + RSS 716 -> 368 MiB. Arm with 1,000-row warmups only (A-baseline): RSS plateau 380-400 MiB (no growth). + So: cyclic garbage holding malloc memory, plus glibc free memory not returned. +- What the garbage is (arms D/E, DEBUG_SAVEALL at idle after 6 bursts): ~9,000 objects; freeing them released + 130 MiB of malloc memory (uordblks 310 -> 180, 265 -> 136) although their Python-visible payload was 20-30 MiB. + Holders: 39 closed client connections (uvicorn HttpToolsProtocol/RequestResponseCycle, asyncio + _SelectorSocketTransport/_SSLProtocolTransport, 78 ssl.MemoryBIO) and 137 refused-item cycles + (QueryServiceError traceback -> `service.py` work:1035 / executor `_one`:301 / `_run_with_policy`:406). +- Cost lines (E4): + 1. `qdl/api_v2/router.py` `_warmup_json_off_loop` / `_warmup_batch_json_off_loop`: the whole warmup body (2-20 MB) + is one `http.response.body` message -> one `transport.write`; asyncio `sslproto._do_write` encrypts it into + the connection's outgoing `ssl.MemoryBIO` in one call. A drained MemoryBIO keeps the capacity of its largest + write until the SSL object is freed: host check, 8 TLS pairs x one 10 MiB write, RSS 27 -> 107 MiB with the + BIOs drained, 37 MiB once the connections were freed. So every client connection pins its largest warmup body, + and a closed connection pins it until a collection frees the connection's cycle. + 2. `qdl/runtime/stable.py` `QUERY_GC_THRESHOLDS = (100_000, 50, 100)` (IDR-114 era, set by + `configure_query_interpreter`): a generation-1 collection needs 50 generation-0 ones; production made 16 in + 32 h, so connection cycles and refused-item cycles that survive one generation-0 collection stay until + restart. Not changed here (it is the GC pause control of IDR-114/F5). + 3. glibc arenas (MALLOC_ARENA_MAX=2) keep freed memory: 150-350 MiB returned only by malloc_trim. + 4. `qdl/query/service.py` assemble_batch kept each refused item's traceback (the IDR-114 residual). + Not the cause (measured): row cache (bounded 20,000), threads (41-55), SQLite (KN backend has no spool), + stall telemetry (bounded deque), refusal log (bounded keys). +- Fix (tested locally + isolated replay): + 1. `SlicedBodyResponse` (`qdl/api_v2/router.py`): warmup and warmup:batch bodies go out as 256 KiB `more_body` + slices; bytes, Content-Length (no chunked encoding) and status unchanged. Bounds each connection's MemoryBIO to + ~256 KiB. + 2. `service.py` assemble_batch: `release_exception_frames(error)` after the typed problem is taken. + 3. `qdl/runtime/memory_maintenance.py` (new), installed by `create_stable_query_app`: when RSS >= soft limit + (`QDL_QUERY_MEMORY_SOFT_LIMIT_MIB`, default 448, 0 disables), the large-warmup lane idle >= 3 s + (`V2QueryService.large_warmup_lane_busy`) and >= 120 s since the last run: one gc.collect() + malloc_trim(0) + on the loop thread, one `qdl.stall kind=memory_maintenance` line. Measured cost of that pair at idle: + gc 133-233 ms, trim 12-55 ms. GC thresholds and the start-up gc.freeze unchanged. +- Results (same harness and bursts, source mounted read-only over the production image): + | arm | RSS burst 1 / 6 / 10 / 12 MiB | uordblks burst 1 -> last | collect frees (idle) | collect+trim RSS | + | --- | --- | --- | --- | --- | + | A-stage production image | 472 / 574 / 652 / 683 | 155 -> 398 | 308 MiB | 716 -> 368 | + | F sliced body + frame release | 395 / 457 / 427 / - | 100 -> 146 | 52 MiB | 438 -> 360 | + | G F + maintenance (512 then) | 391 / 408 / 468 / 458 | 99 -> 161 | 68 MiB | 464 -> 377 | + | H built image 84512ab (default 448) | 361 / 374 / 426 / 445 | 103 -> 116 | 22 MiB | 448 -> 410 | + Garbage after the fix holds no QueryServiceError/work frames (0 vs 137) and only the 256 KiB read buffers of + closed TLS connections (asyncio sslproto). G never reached its 512 MiB limit; default lowered to 448 MiB so the + bounded behaviour is the default one (final proof on the built image below). Client warmup latency (200s): + p50 6.4-8.7 s / p95 9.3-10.2 s on A-stage vs 6.4-8.7 s / 9.0-10.2 s on G (queue-dominated; G overlapped the + affected-suite run on another CPU, so H is the latency comparison). +- Tests (production image 3e9e4d3a, `--network none`, source mounted read-only): new + `tests/test_query_idr129_memory.py` 12 OK (sliced body bytes/length/slices, HTTP identity, TLS MemoryBIO bound, + refused-item no-cycle, maintenance policy, lane-busy signal, lifespan install); negative control: the + refused-item test FAILS on the unfixed service. Affected 37 modules (api_v2, query service, runtime.stable, + warmup, stall telemetry): 552 tests, 1 failure `test_phaseb_stable_edge` catalog L2 count 32 != 34 = the image's + baked config predates dev's IDR-127 catalog (environment, not this change), 36 skipped. +- Final proof on the built image (arm H, 16:28-16:43Z): `scripts/build_qdl_release_images.py --rev 84512ab + --component python --build` -> `qdl-v2-python:2.2.5-84512ab` + `sha256:69304bd478ee7822625c21e8573354c0a24c2baf1f16cc317d216c227f38ef27`, receipt + `~/.local/state/qdl-v2/build-v2.2.5-84512ab/receipt.json` PASS. Same 64-warmup bursts x 12, no source mount, + default soft limit: RSS 197 (start) / 361 (burst 1) / 361-445 over bursts 1-12, malloc in use flat 94-116 MiB + (production image: 155 -> 398). Maintenance ran 3 times: RSS 450 -> 378, 451 -> 374, 454 -> 406 MiB, gc + 128-152 ms, trim 8-12 ms. Client warmup latency (200s) p50 6.6-8.3 s / p95 9.3-10.0 s vs production image + 6.4-8.7 s / 9.3-10.2 s: not worse. Status: implemented, tested locally, isolated-replay measured; not deployed + (ships in v2.2.5; window acceptance = the packet's RSS rule). +- Cleanup: every `qdl-rss-*` container `--rm` and removed (cache, quota, sync, query, driver, unit-test + containers; 0 left), network `qdl-rss-net` removed, no volume created; host lock released 16:43:34Z. Kept: the + image `qdl-v2-python:2.2.5-84512ab` (v2.2.5 candidate) and the evidence dirs above. diff --git a/qdl/api_v2/router.py b/qdl/api_v2/router.py index f8a2ae5b..3f6bd978 100644 --- a/qdl/api_v2/router.py +++ b/qdl/api_v2/router.py @@ -786,6 +786,43 @@ async def _json_off_loop(build) -> JSONResponse: ) +# IDR-129: a large body is handed to the server in slices of this size. asyncio's TLS +# transport encrypts each ``transport.write`` into the connection's OpenSSL output +# buffer (``ssl.MemoryBIO``), and that buffer keeps the capacity of the largest +# single write for the connection's whole life (a drained MemoryBIO that carried +# a 10 MiB write still holds 10 MiB until the SSL object is freed; measured +# 2026-10-19). One write per warmup therefore pinned up to the largest warmup +# body (2-20 MB) per client connection, and a closed connection's objects are +# cyclic garbage that only a rare full collection frees: the isolated 64-warmup +# burst freed 130 MiB of malloc memory from 39 closed connections. Slices bound +# that buffer to ~256 KiB; the bytes, ``Content-Length`` and status are unchanged. +_RESPONSE_SLICE_BYTES = 256 * 1024 + + +class SlicedBodyResponse(Response): + """``Response`` whose body goes to the ASGI server as bounded ``more_body`` slices.""" + + slice_bytes = _RESPONSE_SLICE_BYTES + + async def __call__(self, scope, receive, send) -> None: + if scope["type"] != "http": + await super().__call__(scope, receive, send) + return + await send({"type": "http.response.start", "status": self.status_code, "headers": self.raw_headers}) + body, size, step = self.body, len(self.body), self.slice_bytes + if size <= step: + await send({"type": "http.response.body", "body": body}) + else: + view = memoryview(body) + for start in range(0, size, step): + stop = min(size, start + step) + await send({"type": "http.response.body", "body": bytes(view[start:stop]), + "more_body": stop < size}) + view.release() + if self.background is not None: + await self.background() + + _RENDER_CHUNK_ROWS = 250 # Large responses render on two fixed threads, not on the loop's default # executor (up to cpu_count + 4 threads): each thread that renders a 10k-row @@ -871,7 +908,7 @@ async def _warmup_json_off_loop(build) -> Response: body = await await_in_thread( _RENDER_EXECUTOR, lambda: _render_warmup_result(build()), cold=True ) - return Response(content=body, media_type="application/json") + return SlicedBodyResponse(content=body, media_type="application/json") async def _single_warmup_response( @@ -1036,7 +1073,7 @@ def encode(value): async def _warmup_batch_json_off_loop(request, access, result, requirements) -> Response: body = await await_in_thread(_RENDER_EXECUTOR, lambda: _render_warmup_batch(request, access, result, requirements), cold=True) - return Response(content=body, media_type="application/json") + return SlicedBodyResponse(content=body, media_type="application/json") def _reference_data(result) -> dict: diff --git a/qdl/query/service.py b/qdl/query/service.py index 44d57434..cdc7c7a6 100644 --- a/qdl/query/service.py +++ b/qdl/query/service.py @@ -55,6 +55,7 @@ MarketDataQueryBackend, QualityMetadata, QueryBackendError, + release_exception_frames, CANONICAL_LINEAGE_MISMATCH, CanonicalLineageMismatch, ) @@ -817,6 +818,15 @@ def warmup_batch( ) return BatchQueryResult(request_id, tuple(results)) + def large_warmup_lane_busy(self) -> bool: + """IDR-129: True while the large-warmup lane runs or queues a batch (memory maintenance waits).""" + + admission = getattr(self, "_local_batch_admission", None) + if admission is None: + return False + stats = admission.stats() + return bool(stats.get("active", 0) or stats.get("pending", 0)) + def _local_batch_admission_for( self, requirements: tuple[DataRequirement, ...] = (), ) -> _LocalBatchAdmission: @@ -1114,6 +1124,11 @@ def assemble_batch(executions) -> BatchQueryResult: "warmup batch item failed inside the bounded executor", False, ) + # IDR-129 (the IDR-114 residual): the item error's traceback holds the + # frames of ``work`` and the executor, which hold the error again - + # a cycle per refused item that only a full collection frees. The + # batch keeps the typed problem; the frames are no longer needed. + release_exception_frames(error) results.append( BatchItemResult( requirement.instrument_uid, diff --git a/qdl/runtime/memory_maintenance.py b/qdl/runtime/memory_maintenance.py new file mode 100644 index 00000000..24924077 --- /dev/null +++ b/qdl/runtime/memory_maintenance.py @@ -0,0 +1,175 @@ +"""IDR-129: bounded memory maintenance for a Query replica (full collection + ``malloc_trim``, idle only). + +Purpose: give back memory that the Query's own settings otherwise keep until restart. Measured 2026-10-19 on an +isolated replica of the production image under repeated 64-warmup bursts (``scripts/idr129_query_rss_isolated.py``): + +* Cyclic garbage outlives the bursts. ``QUERY_GC_THRESHOLDS = (100_000, 50, 100)`` (``qdl/runtime/stable.py``, + chosen to stop GC pauses) runs a generation-1 collection only after 50 generation-0 collections and a full one + after 100 of those: production made 16 generation-0 and 0 generation-1 collections in 32 h. Closed client + connections (uvicorn protocol, asyncio TLS transport and their buffers) are such cycles. +* glibc keeps freed arena memory: after the garbage was freed, ``malloc_trim(0)`` returned 150-350 MiB of RSS. + +Policy (one variable, the target is RSS): when the replica's RSS is above a soft limit, the large-warmup lane has +been idle for ``idle_seconds`` and the last maintenance is at least ``min_interval_seconds`` old, run one +``gc.collect()`` and one ``malloc_trim(0)`` on the event-loop thread and record one ``qdl.stall`` line +(``kind=memory_maintenance``). Below the soft limit nothing runs, so a replica that does not accumulate pays no +pause. The thresholds and ``gc.freeze`` of start-up are unchanged. + +Boundary: no I/O besides ``/proc/self/statm`` and the existing stall recorder; never raises into the loop. +""" +from __future__ import annotations + +import asyncio +import ctypes +import gc +import logging +import os +import time +from typing import Callable + +logger = logging.getLogger(__name__) + +SOFT_LIMIT_ENV = "QDL_QUERY_MEMORY_SOFT_LIMIT_MIB" +DEFAULT_SOFT_LIMIT_MIB = 448 +DEFAULT_MIN_INTERVAL_SECONDS = 120.0 +DEFAULT_IDLE_SECONDS = 3.0 +DEFAULT_TICK_SECONDS = 1.0 + + +def _rss_mib() -> float | None: + try: + with open("/proc/self/statm", encoding="ascii") as handle: + return int(handle.read().split()[1]) * os.sysconf("SC_PAGE_SIZE") / 1048576.0 + except (OSError, ValueError, IndexError): + return None + + +def _glibc_malloc_trim() -> Callable[[], int] | None: + try: + libc = ctypes.CDLL("libc.so.6") + trim = libc.malloc_trim + except (OSError, AttributeError): + return None + trim.argtypes = [ctypes.c_size_t] + trim.restype = ctypes.c_int + return lambda: int(trim(0)) + + +def soft_limit_from_environment(environ=os.environ) -> float: + raw = environ.get(SOFT_LIMIT_ENV, "").strip() + if not raw: + return float(DEFAULT_SOFT_LIMIT_MIB) + value = float(raw) + if value < 0: + raise ValueError(f"{SOFT_LIMIT_ENV} must be >= 0 (0 disables maintenance)") + return value + + +class MemoryMaintenance: + """Decides and runs one maintenance step per ``tick`` (see the module docstring).""" + + def __init__( + self, + *, + busy: Callable[[], bool], + soft_limit_mib: float = DEFAULT_SOFT_LIMIT_MIB, + min_interval_seconds: float = DEFAULT_MIN_INTERVAL_SECONDS, + idle_seconds: float = DEFAULT_IDLE_SECONDS, + record: Callable[[dict], None] | None = None, + rss: Callable[[], float | None] = _rss_mib, + collect: Callable[[], int] = gc.collect, + trim: Callable[[], int] | None = None, + clock: Callable[[], float] = time.monotonic, + wall_ms: Callable[[], int] = lambda: time.time_ns() // 1_000_000, + ) -> None: + if soft_limit_mib < 0 or min_interval_seconds < 0 or idle_seconds < 0: + raise ValueError("memory maintenance bounds must be >= 0") + self.soft_limit_mib = soft_limit_mib + self.min_interval_seconds = min_interval_seconds + self.idle_seconds = idle_seconds + self._busy = busy + self._record = record + self._rss = rss + self._collect = collect + self._trim = trim if trim is not None else (_glibc_malloc_trim() or (lambda: 0)) + self._clock = clock + self._wall_ms = wall_ms + self._last_busy = clock() + self._last_run: float | None = None + self.runs = 0 + + def tick(self) -> dict | None: + """One decision; returns the record of a maintenance step or ``None``.""" + + if self.soft_limit_mib == 0: + return None + now = self._clock() + if self._busy(): + self._last_busy = now + return None + if now - self._last_busy < self.idle_seconds: + return None + if self._last_run is not None and now - self._last_run < self.min_interval_seconds: + return None + before = self._rss() + if before is None or before < self.soft_limit_mib: + return None + self._last_run = now + started = time.perf_counter() + collected = self._collect() + collected_at = time.perf_counter() + self._trim() + finished = time.perf_counter() + self.runs += 1 + record = { + "kind": "memory_maintenance", "rss_before_mib": round(before, 1), + "rss_after_mib": None if (after := self._rss()) is None else round(after, 1), + "collected": collected, "gc_ms": round((collected_at - started) * 1000.0, 3), + "trim_ms": round((finished - collected_at) * 1000.0, 3), + "soft_limit_mib": self.soft_limit_mib, "wall_ms": self._wall_ms(), + } + if self._record is not None: + try: + self._record(record) + except Exception: # telemetry never breaks maintenance + pass + return record + + +async def run_memory_maintenance(maintenance: MemoryMaintenance, *, tick_seconds: float = DEFAULT_TICK_SECONDS): + """Tick on the event loop until cancelled (a collection holds the GIL on any thread).""" + + while True: + await asyncio.sleep(tick_seconds) + try: + maintenance.tick() + except Exception: # pragma: no cover - defensive: maintenance never stops the replica + logger.exception("query memory maintenance tick failed") + + +def install_query_memory_maintenance(app, service, *, recorder=None) -> MemoryMaintenance | None: + """Start the maintenance task with the app; ``None`` when disabled (soft limit 0).""" + + soft_limit = soft_limit_from_environment() + if soft_limit == 0: + return None + busy = getattr(service, "large_warmup_lane_busy", None) + maintenance = MemoryMaintenance( + busy=busy if callable(busy) else (lambda: False), + soft_limit_mib=soft_limit, + record=recorder.record if recorder is not None else None, + ) + tasks: list[asyncio.Task] = [] + + @app.on_event("startup") + async def start_memory_maintenance() -> None: + tasks.append(asyncio.get_running_loop().create_task( + run_memory_maintenance(maintenance), name="qdl-memory-maintenance")) + + @app.on_event("shutdown") + async def stop_memory_maintenance() -> None: + for task in tasks: + task.cancel() + + app.state.memory_maintenance = maintenance + return maintenance diff --git a/qdl/runtime/stable.py b/qdl/runtime/stable.py index 72b5fc2f..123da23a 100644 --- a/qdl/runtime/stable.py +++ b/qdl/runtime/stable.py @@ -38,7 +38,12 @@ MeasuredRuntimeReadiness, ) from qdl.runtime.stable_catalog import StableSourceCatalog -from qdl.runtime.stall_telemetry import install_gc_pause_telemetry, install_request_stall_telemetry +from qdl.runtime.memory_maintenance import install_query_memory_maintenance +from qdl.runtime.stall_telemetry import ( + install_gc_pause_telemetry, + install_request_stall_telemetry, + process_stall_recorder, +) from qdl.runtime.stable_capacity import STABLE_SPOOL_PHYSICAL_PARTITION_WINDOW from qdl.runtime.stable_deployment import ( StableAcquisitionPlan, @@ -807,6 +812,9 @@ def create_stable_query_app(config: StableRuntimeConfig | None = None) -> FastAP app.state.stable_query_backend = query_backend # IDR-114: slow requests and event-loop lag on the replica itself (observation only). install_request_stall_telemetry(app) + # IDR-129: full collection + malloc_trim when RSS is above the soft limit and the + # large-warmup lane is idle (QDL_QUERY_MEMORY_SOFT_LIMIT_MIB, 0 disables). + install_query_memory_maintenance(app, service, recorder=process_stall_recorder()) read_view_secret_file = os.environ.get(READ_VIEW_SECRET_FILE_ENV, "").strip() if kn_backend is not None and read_view_secret_file: # D29: the native Stream's GetSnapshot/GetFeedStatus read view. diff --git a/scripts/idr129_query_rss_isolated.py b/scripts/idr129_query_rss_isolated.py new file mode 100644 index 00000000..79838527 --- /dev/null +++ b/scripts/idr129_query_rss_isolated.py @@ -0,0 +1,775 @@ +#!/usr/bin/env python3 +"""IDR-129: isolated Query replica under repeated warmup bursts, for retained-RSS measurement. + +Purpose: find what keeps a Query replica's RSS after the TS writer's warmup burst (production query_kn_1 reached +2045 MiB of 2 GiB over 32 h, one step per burst) and prove a fix, where the product runs: the Query container of +the production image over HTTPS mTLS + JWT, driven by the SDK's own redirect/retry pattern (64 concurrent +1,000-row BAR warmups per burst, ``X-QDL-Replica-Redirect: 1`` then a queued retry on ``REPLICA_LANE_BUSY``, +then the cold-read retry on a retryable 429). + +Topology (prefix ``qdl-rss-``, label ``qdl.idr129.rss=true``, every container ``--rm``, one ``--internal`` network): + qdl-rss-cache private Redis (stack Redis digest, tmpfs): a read-only COPY of the market-cache keys of the + burst's products (``ptr``, ``src``, ``bm``, the ``b``/``bd``/``bs`` buckets of the last + rows and the bars ``ckpt`` hashes). Copied once by ``setup`` with DUMP/HMGET/HGETALL only + (read-only commands) from the production market cache, then RESTOREd here. + qdl-rss-quota private quota/lease Redis (the production replica's quota Redis is never used). + qdl-rss-query the production Query container spec (env, read-only mounts, uid 10001) except: market cache + and quota URLs, a scratch state dir, instance id, ``--cpus 1.0 --memory 2g``; optional source + mounted read-only over ``/app/qdl`` (``--src``) or another image (``--image``). A + ``sitecustomize`` probe (``/rssprobe``) answers commands from the driver through a shared + scratch dir: stats (RSS, RssAnon, threads, gc counts and generation sizes, mallinfo2, row + cache), ``gc.collect``, ``malloc_trim(0)``, tracemalloc snapshots and diffs. + qdl-rss-driver SDK-shaped burst driver (TS identity, plus the alpha identities for the other products). +Production is read-only: the copy reads the market cache once; nothing is written to a production Redis, +container, network alias, volume or file. Run under the host measurement lock (AGENTS.md 34a). + +Usage (host python): + idr129_query_rss_isolated.py setup --output DIR (network, private cache copy, quota Redis) + idr129_query_rss_isolated.py arm --output DIR --name A [--src DIR | --image REF] [--bursts N] [--tracemalloc F] + [--env K=V ...] (one replica + driver; replica removed after) + idr129_query_rss_isolated.py cleanup (removes every qdl-rss-* container and the network) +""" +from __future__ import annotations + +import argparse +import json +import os +from pathlib import Path +import subprocess +import sys +import time + +PREFIX = "qdl-rss-" +LABEL = "qdl.idr129.rss=true" +NETWORK = PREFIX + "net" +STACK = "qdl_v2_stable_candidate" +PROD_QUERY = f"{STACK}-query_kn_1-1" +PROD_INTERNAL = f"{STACK}_stable_internal" +REDIS_IMAGE = "sha256:dfa18828cbc07b3ae6a95ec7343f6c214fdee2d836197b4be8e9904420762cd8" +QUERY_IMAGE = "sha256:3e9e4d3add0f35c5d4b4796c785a35b466c326d3328184f396182b5f51367fff" +LOCK = Path.home() / ".local/state/host-measurement-window.lock" +MOUNTS = ("/runtime", "/session-liveness", "/consumer-realms", "/kn", "/stable-certs") +IDENTITIES = Path.home() / ".local/state/qdl-v2/r135-local-batch-fair-e905747-20260920T135150Z/identities" +SLICES = 64 +ROWS = 1000 # TS BAR warmup rows (manifest warmup_limit) +COPY_ROWS = 10_000 # rows retained in the private copy per product (largest warmup of the stage mix) +STAGE_ALPHA_ROWS = (2500, 5000, 2500, 5000, 2500, 5000, 2500, 10_000) # stage-100 cold role + KN-4 ladder + +# Requirements of the burst: every TS BAR (1m, 1,000 rows), then alpha BAR products of the most retained +# intervals until SLICES products exist (each must retain >= ROWS rows in the cache, checked by the copy). +PLAN = r''' +import json, sys, yaml +manifests = { + "trading-system.paper.stable": "/consumer-realms/trading-system.paper.stable-r11.yaml", + "alpha.binance.paper.stable": "/consumer-realms/alpha.binance.paper.stable-r16.yaml", + "alpha.okx.paper.stable": "/consumer-realms/alpha.okx.paper.stable-r15.yaml", +} +rows = [] +for consumer, path in manifests.items(): + spec = yaml.safe_load(open(path))["spec"] + for item in spec["requirements"]: + if item["feed"] == "BAR" and item.get("interval") in ("1m", "3m", "5m", "15m", "30m", "1h"): + rows.append({"consumer": consumer, "requirement": item}) +order = {"1m": 0, "3m": 1, "5m": 2, "15m": 3, "30m": 4, "1h": 5} +rows.sort(key=lambda r: (r["consumer"] != "trading-system.paper.stable", order[r["requirement"]["interval"]])) +json.dump(rows, sys.stdout) +''' + +COPY = r''' +"""Read-only copy of the burst products' market-cache keys (DUMP/HMGET/HGETALL) into the private cache.""" +import json, os, sys +import redis +from qdl.adapters.intervals import canonical_interval_ms +from qdl.query import DataRequirement +from qdl.projection.kn_state_codec import state_partition +from qdl.runtime.kn_bar_readback import BUCKET_OPENS, binding_product_key +from qdl.runtime.kn_market_cache import KnMarketCacheReader +from qdl.runtime.stable_catalog import StableSourceCatalog + +plan = json.load(open(sys.argv[1])) +need, rows_needed = int(sys.argv[2]), int(sys.argv[3]) +src = redis.Redis.from_url(os.environ["SRC_URL"], socket_timeout=5) +dst = redis.Redis.from_url(os.environ["DST_URL"], socket_timeout=5) +catalog = StableSourceCatalog.load("/runtime/stable-source-bindings.yaml") +reader = KnMarketCacheReader(src, "paper") +chosen, keys, size, ckpts = [], 0, 0, set() + +def copy(key): + global keys, size + blob = src.dump(key) + if blob is None: + return False + dst.restore(key, 0, blob, replace=True) + keys += 1 + size += len(blob) + return True + +def copy_product(entry, *, tail_only=False, generations={}): + item = dict(entry["requirement"]) + try: + binding = catalog.binding_for(DataRequirement.from_mapping({**item, "warmup_limit": 0})) + except Exception: + return None + lpk = binding_product_key(binding, "paper") + code = lpk.encode() + ready = src.hmget(reader.pointer_key(lpk), "ready")[0] + if ready is None: + return None + generation = int(ready) + meta = {k.decode(): v.decode() for k, v in src.hgetall(f"kn3:paper:bm:{generation}:{code}").items()} + if int(meta.get("rows") or 0) < rows_needed + 5 or not meta.get("last"): + return None + step = canonical_interval_ms(binding.interval) + last = int(meta["last"]) + tail_only = tail_only and generations.get(code) == generation + generations[code] = generation + first_needed = last - (2 * BUCKET_OPENS if tail_only else rows_needed + 200) * step + for key in (reader.pointer_key(lpk), f"kn3:paper:src:{code}", f"kn3:paper:bm:{generation}:{code}"): + copy(key) + for bucket in range(first_needed // (BUCKET_OPENS * step), last // (BUCKET_OPENS * step) + 1): + for kind in ("b", "bd", "bs"): + copy(f"kn3:paper:{kind}:{generation}:{code}:{bucket}") + ckpts.add(reader.checkpoint_key(reader.bars_topic, state_partition(lpk, reader.bars_partitions))) + return {**entry, "binding_id": binding.binding_id, "retained_rows": int(meta["rows"])} + +sync_seconds = float(os.environ.get("SYNC_SECONDS", "0")) +if sync_seconds: + # Keep the copy as fresh as the production cache: re-copy each product's tail every sync_seconds (read-only + # source commands only) until the deadline; a generation change re-copies the whole window. + import time + deadline = time.monotonic() + float(os.environ["SYNC_DEADLINE_SECONDS"]) + rounds = 0 + while time.monotonic() < deadline: + for entry in plan: + copy_product(entry, tail_only=rounds > 0) + for key in sorted(ckpts): + copy(key) + rounds += 1 + print(json.dumps({"round": rounds, "keys": keys, "dump_bytes": size}), flush=True) + time.sleep(sync_seconds) + sys.exit(0) +for entry in plan: + if len(chosen) >= need: + break + copied = copy_product(entry) + if copied is not None: + chosen.append(copied) +for key in sorted(ckpts): + copy(key) +print(json.dumps({"products": chosen, "keys": keys, "dump_bytes": size})) +''' + +PROBE = r''' +"""IDR-129 RSS probe (isolated replica only): answers driver commands written to /probe; no payloads.""" +import collections, ctypes, gc, json, os, sys, threading, time, tracemalloc + +_snaps = {} + +class _Mallinfo2(ctypes.Structure): + _fields_ = [(n, ctypes.c_size_t) for n in ("arena", "ordblks", "smblks", "hblks", "hblkhd", "usmblks", + "fsmblks", "uordblks", "fordblks", "keepcost")] + +try: + _libc = ctypes.CDLL("libc.so.6") + _libc.mallinfo2.restype = _Mallinfo2 + _libc.malloc_trim.argtypes = [ctypes.c_size_t] +except Exception: + _libc = None + +def _status(): + out = {} + with open("/proc/self/status") as handle: + for line in handle: + key, _, value = line.partition(":") + if key in ("VmRSS", "RssAnon", "RssFile", "VmHWM", "Threads"): + out[key] = int(value.split()[0]) + return out + +_caches = [] + +def _row_caches(): + # The backend is built before gc.freeze(), so gc.get_objects() never lists it: instances are registered here. + return [cache.stats() for cache in _caches] + +def _register_row_caches(): + try: + from qdl.query import row_cache + except Exception: + return + original = row_cache.BoundedRowCache.__init__ + def init(self, *args, **kwargs): + original(self, *args, **kwargs) + _caches.append(self) + row_cache.BoundedRowCache.__init__ = init + +def stats(types=False): + s = _status() + record = {"wall_ms": time.time_ns() // 1_000_000, "rss_mib": round(s["VmRSS"] / 1024, 1), + "rss_anon_mib": round(s["RssAnon"] / 1024, 1), "hwm_mib": round(s["VmHWM"] / 1024, 1), + "threads": s["Threads"], "thread_names": collections.Counter( + t.name.rstrip("0123456789_-") for t in threading.enumerate()).most_common(), + "gc_count": gc.get_count(), "gc_collections": [g["collections"] for g in gc.get_stats()], + "gen_sizes": [len(gc.get_objects(g)) for g in range(3)], "frozen": gc.get_freeze_count(), + "allocated_blocks": sys.getallocatedblocks(), "row_caches": _row_caches()} + if _libc is not None: + m = _libc.mallinfo2() + record["mallinfo2_mib"] = {n: round(getattr(m, n) / 1048576, 1) for n in ("arena", "hblkhd", "uordblks", + "fordblks", "keepcost")} + if tracemalloc.is_tracing(): + current, peak = tracemalloc.get_traced_memory() + record["tracemalloc_mib"] = [round(current / 1048576, 1), round(peak / 1048576, 1)] + if types: + record["types_gen2"] = collections.Counter(type(o).__qualname__ for o in gc.get_objects(2)).most_common(25) + return record + +def run(command): + action = command["action"] + if action == "stats": + return stats(types=command.get("types", False)) + if action == "collect": + before = stats() + started = time.perf_counter() + collected = gc.collect() + return {"before": before, "collected": collected, "ms": round((time.perf_counter() - started) * 1e3, 1), + "after": stats()} + if action == "trim": + before = stats() + started = time.perf_counter() + released = _libc.malloc_trim(0) if _libc is not None else None + return {"before": before, "released": released, "ms": round((time.perf_counter() - started) * 1e3, 1), + "after": stats()} + if action == "tm_snapshot": + _snaps[command["name"]] = tracemalloc.take_snapshot().filter_traces( + (tracemalloc.Filter(False, tracemalloc.__file__), tracemalloc.Filter(False, __file__))) + return {"name": command["name"]} + if action == "tm_diff": + old, new = _snaps[command["old"]], _snaps[command["new"]] + key = command.get("key", "lineno") + diff = new.compare_to(old, key) + top = [{"size_diff_kib": round(d.size_diff / 1024, 1), "size_kib": round(d.size / 1024, 1), + "count_diff": d.count_diff, "where": [f"{f.filename.rsplit('/site-packages/', 1)[-1].rsplit('/app/', 1)[-1]}:{f.lineno}" + for f in d.traceback[:8]]} for d in diff[:40]] + live = [{"size_kib": round(s.size / 1024, 1), "count": s.count, + "where": [f"{f.filename.rsplit('/site-packages/', 1)[-1].rsplit('/app/', 1)[-1]}:{f.lineno}" + for f in s.traceback[:8]]} for s in new.statistics(key)[:30]] + return {"diff": top, "live": live} + if action == "saveall_collect": + before = stats() + gc.set_debug(gc.DEBUG_SAVEALL) + started = time.perf_counter() + collected = gc.collect() + ms = round((time.perf_counter() - started) * 1e3, 1) + gc.set_debug(0) + garbage = list(gc.garbage) + gc.garbage.clear() + summary = _garbage_summary(garbage) + del garbage + # The saved garbage is still cyclic: one more (plain) collection frees it, which is what it was holding. + held = stats() + freed = gc.collect() + return {"before": before, "collected": collected, "ms": ms, "summary": summary, "held": held, + "freed": freed, "after": stats()} + if action == "holders": + return holders(command.get("min_bytes", 262144)) + raise ValueError(action) + +def _bulk(value): + """Rough payload size of a local: bytes/str length, or element count x 1 KiB for long sequences/dicts, + or the item count of a history-like object (``items``/``history.items``).""" + if isinstance(value, (bytes, bytearray, str)): + return len(value) + if isinstance(value, (list, tuple, dict)): + return len(value) * 1024 if len(value) >= 64 else sum(_bulk(v) for v in (value.values() if isinstance(value, dict) else value) if isinstance(v, (bytes, bytearray)) ) + for attr in ("items", "body"): + inner = getattr(value, attr, None) + if isinstance(inner, (bytes, bytearray)): + return len(inner) + if isinstance(inner, tuple): + return len(inner) * 1024 + history = getattr(value, "history", None) or getattr(value, "result", None) + if history is not None and history is not value: + items = getattr(history, "items", None) or getattr(getattr(history, "history", None), "items", None) + if isinstance(items, tuple): + return len(items) * 1024 + return 0 + +def _garbage_summary(objects): + """Cyclic garbage by type, frame, exception and coroutine, plus the bytes/str payload one level below.""" + types, frames, excs, coros, leaf_types, big_locals = (collections.Counter() for _ in range(6)) + leaf = 0 + for obj in objects: + name = f"{type(obj).__module__}.{type(obj).__qualname__}" + types[name] += 1 + if type(obj).__name__ == "frame": + code = obj.f_code + where = f"{code.co_filename.rsplit('/app/', 1)[-1].rsplit('/site-packages/', 1)[-1]}:{code.co_qualname}:{obj.f_lineno}" + frames[where] += 1 + try: + for local, value in obj.f_locals.items(): + size = _bulk(value) + if size >= 65536: + big_locals[f"{where} {local}={type(value).__qualname__}"] += size + except Exception: + pass + elif isinstance(obj, BaseException): + excs[name] += 1 + elif type(obj).__name__ == "coroutine": + coros[obj.__qualname__] += 1 + for attr in ("_write_backlog", "_buffer"): # asyncio sslproto / selector transport pending writes + pending = getattr(obj, attr, None) if not isinstance(obj, (dict, tuple, list)) else None + if isinstance(pending, (collections.deque, bytearray)): + size = len(pending) if isinstance(pending, bytearray) else sum( + (x.nbytes if isinstance(x, memoryview) else len(x)) for x in pending + if isinstance(x, (bytes, bytearray, memoryview))) + leaf_types[f"pending {attr} of {type(obj).__qualname__}"] += size + for ref in gc.get_referents(obj): + if isinstance(ref, (bytes, bytearray, str, memoryview)): + size = ref.nbytes if isinstance(ref, memoryview) else sys.getsizeof(ref) + leaf += size + leaf_types[f"{type(ref).__name__} via {type(obj).__qualname__}"] += size + return {"objects": len(objects), "leaf_mib_one_level": round(leaf / 1048576, 1), + "leaf_by_holder_mib": [(k, round(v / 1048576, 2)) for k, v in leaf_types.most_common(15)], + "types": types.most_common(30), "frames": frames.most_common(40), "exceptions": excs.most_common(15), + "coroutines": coros.most_common(15), + "big_locals_mib": [(k, round(v / 1048576, 2)) for k, v in big_locals.most_common(20)]} + +def _describe(obj): + kind = f"{type(obj).__module__}.{type(obj).__qualname__}" + if type(obj).__name__ == "frame": + code = obj.f_code + return f"frame {code.co_filename.rsplit('/app/', 1)[-1].rsplit('/site-packages/', 1)[-1]}:{code.co_qualname}:{obj.f_lineno}" + if type(obj).__name__ in ("coroutine", "Task"): + return f"{kind} {getattr(obj, '__qualname__', '') or getattr(obj, 'get_name', lambda: '')()}" + if isinstance(obj, BaseException): + return f"exception {kind}" + return kind + +def holders(min_bytes): + """Large bytes/bytearray/str reachable one level below a gc-tracked container, with three levels of + referrers (types, frames and coroutine names only; never the payload).""" + gc.collect(0) + seen, found = set(), [] + objects = gc.get_objects() + for container in objects: + try: + refs = gc.get_referents(container) + except Exception: + continue + for ref in refs: + if isinstance(ref, (bytes, bytearray, str)) and len(ref) >= min_bytes and id(ref) not in seen: + seen.add(id(ref)) + chain, level = [], [container] + for _ in range(3): + names = collections.Counter(_describe(o) for o in level) + chain.append(names.most_common(6)) + nxt = [] + for o in level[:4]: + nxt.extend(r for r in gc.get_referrers(o) + if r is not objects and r is not level and r is not nxt and r is not refs + and not (type(r) is list and len(r) >= 1000)) + level = nxt[:8] + if not level: + break + found.append({"bytes": len(ref), "type": type(ref).__name__, "chain": chain}) + del objects + return {"count": len(found), "total_mib": round(sum(f["bytes"] for f in found) / 1048576, 1), + "items": found[:25]} + +def _loop(): + root = "/probe" + while True: + try: + for name in sorted(os.listdir(root)): + if name.startswith("cmd-") and name.endswith(".json"): + path = os.path.join(root, name) + command = json.load(open(path)) + os.unlink(path) + try: + reply = {"ok": True, "result": run(command)} + except Exception as error: + reply = {"ok": False, "error": repr(error)} + tmp = os.path.join(root, "tmp-" + name) + with open(tmp, "w") as handle: + json.dump(reply, handle) + os.rename(tmp, os.path.join(root, "out-" + name[4:])) + except FileNotFoundError: + pass + time.sleep(0.2) + +if os.getpid() == 1 and os.path.isdir("/probe"): + frames = int(os.environ.get("QDL_RSS_TRACEMALLOC", "0") or 0) + if frames: + tracemalloc.start(frames) + _register_row_caches() + threading.Thread(target=_loop, name="rss-probe", daemon=True).start() +''' + +DRIVER = r''' +"""IDR-129 burst driver: SDK-shaped concurrent BAR warmups against one isolated Query replica.""" +import asyncio, json, os, statistics, sys, time +import httpx +from qdl_sdk.credentials import RotatingJwtCredentialProvider +from qdl_sdk.errors import DataLayerError +from qdl_sdk.models import (BarRevisionPolicy, DataRequirement, Feed, GapPolicy, Grade, RecoveryPolicy, + StalePolicy) +from qdl_sdk.tls import WorkloadTlsConfig +from qdl_sdk.transport import RestQueryTransport, REPLICA_LANE_BUSY + +ISSUER, AUDIENCE = "https://identity.qdl.stable.internal", "qdl-v2-stable" +ROLES = ("historical_reader", "market_data_reader", "stream_consumer") +IDS = {"trading-system.paper.stable": ("trading-system", "stable-trading-system-rs256-v1"), + "alpha.binance.paper.stable": ("alpha-binance", "stable-alpha-binance-rs256-v1"), + "alpha.okx.paper.stable": ("alpha-okx", "stable-alpha-okx-rs256-v1")} +SUBJECTS = {"trading-system.paper.stable": ("spiffe://qdl/paper/trading-system-stable", 11), + "alpha.binance.paper.stable": ("spiffe://qdl/paper/alpha-binance-stable", 16), + "alpha.okx.paper.stable": ("spiffe://qdl/paper/alpha-okx-stable", 15)} +cfg = json.load(open("/work/driver.json")) +seq = 0 + +def probe(action, **kw): + global seq + seq += 1 + name = f"{seq:05d}" + tmp = f"/probe/tmpcmd-{name}" + with open(tmp, "w") as handle: + json.dump({"action": action, **kw}, handle) + os.rename(tmp, f"/probe/cmd-{name}.json") + deadline = time.monotonic() + 300 + while time.monotonic() < deadline: + out = f"/probe/out-{name}.json" + if os.path.exists(out): + reply = json.load(open(out)) + os.unlink(out) + return reply + time.sleep(0.1) + return {"ok": False, "error": "probe timeout"} + +def transport(consumer): + slug, key_id = IDS[consumer] + subject, revision = SUBJECTS[consumer] + base = f"/identities/{slug}" + tls = WorkloadTlsConfig(f"{base}/ca.crt", f"{base}/client.crt", f"{base}/client.key") + cred = RotatingJwtCredentialProvider(private_key_file=f"/identities/{slug}-jwt/private.key", key_id=key_id, + algorithm="RS256", issuer=ISSUER, audience=AUDIENCE, subject=subject, environment="paper", roles=ROLES, + venues=("BINANCE", "OKX"), consumer_manifest_revision=revision, lifetime_seconds=300, + refresh_before_seconds=60) + return RestQueryTransport(cfg["url"], timeout_seconds=10.0, credential_provider=cred, tls=tls) + +async def slice_warmup(t, consumer, requirement, out): + """SDK 2.4 shape: up to 3 cold-read attempts per round, each redirect=1 then a queued retry; rounds = the TS + bridge reconnect loop (1 s backoff).""" + started = time.perf_counter() + for round_ in range(cfg["rounds"]): + for attempt in range(3): + error = None + for redirect in (True, False): + t0 = time.perf_counter() + try: + payload = await t.warmup(requirement, consumer_id=consumer, replica_redirect=redirect) + except DataLayerError as caught: + out["requests"].append(["429" if caught.code == "RATE_LIMITED" else caught.code, + caught.reason, round((time.perf_counter() - t0) * 1e3)]) + error = caught + if redirect and caught.code == "RATE_LIMITED" and caught.reason == REPLICA_LANE_BUSY: + continue + break + except httpx.HTTPError as caught: + out["requests"].append(["TRANSPORT", type(caught).__name__, round((time.perf_counter() - t0) * 1e3)]) + error = caught + break + out["requests"].append(["200", len(payload.get("data", ())), round((time.perf_counter() - t0) * 1e3)]) + out["ok"].append(round((time.perf_counter() - started) * 1e3)) + return + if not isinstance(error, DataLayerError) or error.code != "RATE_LIMITED" or not error.retryable: + break + delay = max(250 * (attempt + 1), error.retry_after_ms or 0) + await asyncio.sleep(delay / 1000) + await asyncio.sleep(1.0) + out["failed"] += 1 + +async def main(): + transports = {c: transport(c) for c in IDS} + enums = {"feed": Feed, "consumer_grade": Grade, "stale_policy": StalePolicy, "gap_policy": GapPolicy, + "recovery": RecoveryPolicy, "bar_revision_policy": BarRevisionPolicy, "event_recency_policy": StalePolicy} + reqs = [] + for p in cfg["products"]: + fields = {k: (enums[k](v) if k in enums and v is not None else v) for k, v in p["requirement"].items()} + reqs.append((p["consumer"], DataRequirement(**{**fields, "warmup_limit": p["rows"]}))) + ready_deadline = time.monotonic() + 600 + while True: # the replica serves HTTPS only after start-up (slower under tracemalloc) + try: + await transports["trading-system.paper.stable"]._client.get("/health/live") + break + except httpx.HTTPError: + if time.monotonic() > ready_deadline: + raise + await asyncio.sleep(2) + await asyncio.sleep(5) + record = {"baseline": probe("stats", types=True), "bursts": []} + if cfg.get("tracemalloc"): + probe("tm_snapshot", name="start") + for burst in range(cfg["bursts"]): + out = {"requests": [], "ok": [], "failed": 0} + t0 = time.perf_counter() + await asyncio.gather(*(slice_warmup(transports[c], c, r, out) for c, r in reqs)) + wall = round(time.perf_counter() - t0, 1) + await asyncio.sleep(cfg["settle_s"]) + statuses = {} + for status, _, _ in out["requests"]: + statuses[status] = statuses.get(status, 0) + 1 + ok200 = [ms for s, _, ms in out["requests"] if s == "200"] + entry = {"burst": burst + 1, "wall_s": wall, "statuses": statuses, "failed_slices": out["failed"], + "slice_ok_ms_p50": statistics.median(out["ok"]) if out["ok"] else None, + "slice_ok_ms_max": max(out["ok"], default=None), + "req200_ms_p50": statistics.median(ok200) if ok200 else None, + "req200_ms_p95": sorted(ok200)[int(0.95 * (len(ok200) - 1))] if ok200 else None, + "rows": sorted({n for s, n, _ in out["requests"] if s == "200"}), + "stats": probe("stats")} + if burst == 0 and cfg.get("tracemalloc"): + probe("tm_snapshot", name="after1") + record["bursts"].append(entry) + r = entry["stats"].get("result", {}) + print(json.dumps({k: entry[k] for k in ("burst", "wall_s", "statuses", "failed_slices", "req200_ms_p50")} + | {"rss_mib": r.get("rss_mib"), "threads": r.get("threads"), "gen": r.get("gen_sizes")}), + flush=True) + await asyncio.sleep(cfg["gap_s"]) + record["idle"] = probe("stats", types=True) + record["holders"] = probe("holders") + if cfg.get("tracemalloc"): + probe("tm_snapshot", name="end") + record["tm_diff_after1_end"] = probe("tm_diff", old="after1", new="end") + record["tm_diff_start_end_tb"] = probe("tm_diff", old="start", new="end", key="traceback") + if cfg.get("tracemalloc"): + probe("tm_snapshot", name="before_collect") + record["collect"] = probe("saveall_collect") + if cfg.get("tracemalloc"): + probe("tm_snapshot", name="after_collect") + record["tm_freed_by_collect"] = probe("tm_diff", old="before_collect", new="after_collect", key="traceback") + record["trim"] = probe("trim") + json.dump(record, open("/work/result.json", "w"), indent=1) + for t in transports.values(): + await t.close() + +asyncio.run(main()) +''' + + +def sh(*args, check=True, timeout=180): + result = subprocess.run(list(args), capture_output=True, text=True, timeout=timeout) + if check and result.returncode != 0: + raise RuntimeError(f"{args[:4]} failed: {result.stderr.strip()[:600]}") + return result.stdout.strip() + + +def require_lock(owner: str): + try: + lock = json.loads(LOCK.read_text()) + except (OSError, ValueError): + raise SystemExit("host measurement lock is absent: acquire it first (AGENTS.md 34a)") + if lock.get("owner") != owner: + raise SystemExit(f"host measurement lock is held by {lock.get('owner')}, not {owner}") + + +def prod_spec(): + raw = json.loads(sh("docker", "inspect", PROD_QUERY))[0] + mounts = {m["Destination"]: m["Source"] for m in raw["Mounts"] if m["Destination"] in MOUNTS} + if set(mounts) != set(MOUNTS): + raise RuntimeError(f"production Query mounts differ: {sorted(mounts)}") + env = dict(item.split("=", 1) for item in raw["Config"]["Env"]) + return raw["Image"], env, mounts, raw["Config"]["Cmd"] + + +def ro_mounts(mounts: dict) -> list[str]: + return [a for dst, src in sorted(mounts.items()) for a in ("--mount", f"type=bind,src={src},dst={dst},readonly")] + + +def setup(args) -> int: + require_lock(args.lock_owner) + output = args.output.resolve() + output.mkdir(mode=0o700, parents=True, exist_ok=False) + image, _env, mounts, _cmd = prod_spec() + if image != QUERY_IMAGE: + raise SystemExit(f"production Query image is {image[:19]}, expected {QUERY_IMAGE[:19]}") + work = output / "work" + work.mkdir(mode=0o777) + os.chmod(work, 0o777) + sh("docker", "network", "create", "--internal", "--label", LABEL, NETWORK) + for name, memory in (("cache", "768m"), ("quota", "128m")): + sh("docker", "run", "-d", "--rm", "--name", PREFIX + name, "--label", LABEL, "--network", NETWORK, + "--cpus", "0.5", "--memory", memory, "--read-only", "--tmpfs", "/data:size=64m", + REDIS_IMAGE, "redis-server", "--save", "", "--appendonly", "no", "--maxmemory", + "640mb" if name == "cache" else "96mb") + plan = sh("docker", "run", "--rm", "--network", "none", "--user", "10001:10001", "--read-only", + "--cpus", "0.5", *ro_mounts({"/consumer-realms": mounts["/consumer-realms"]}), + QUERY_IMAGE, "python", "-c", PLAN) + (work / "plan.json").write_text(plan) + os.chmod(work / "plan.json", 0o644) + name = PREFIX + "copy" + sh("docker", "create", "--rm", "--name", name, "--label", LABEL, "--network", NETWORK, "--user", "10001:10001", + "--read-only", "--cpus", "0.5", "--memory", "512m", "-e", "PYTHONPATH=/app", + "-e", "SRC_URL=redis://market_cache:6379/0", "-e", f"DST_URL=redis://{PREFIX}cache:6379/0", + *ro_mounts({"/runtime": mounts["/runtime"]}), + "--mount", f"type=bind,src={work},dst=/work,readonly", + QUERY_IMAGE, "python", "-c", COPY, "/work/plan.json", str(args.slices), str(COPY_ROWS)) + sh("docker", "network", "connect", PROD_INTERNAL, name) + started = time.time() + copied = subprocess.run(["docker", "start", "-a", name], capture_output=True, text=True, timeout=600) + if copied.returncode != 0: + raise SystemExit(f"copy failed: {copied.stderr[-1500:]}") + result = json.loads(copied.stdout.strip().splitlines()[-1]) + result["copy_seconds"] = round(time.time() - started, 1) + result["copied_utc"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + (output / "products.json").write_text(json.dumps(result, indent=1) + "\n") + print(json.dumps({"products": len(result["products"]), "keys": result["keys"], "dump_bytes": result["dump_bytes"], + "copy_seconds": result["copy_seconds"]})) + return 0 if len(result["products"]) == args.slices else 1 + + +def arm(args) -> int: + require_lock(args.lock_owner) + output = args.output.resolve() + products = json.loads((output / "products.json").read_text())["products"] + image, env, mounts, cmd = prod_spec() + run_image = args.image or QUERY_IMAGE + arm_dir = output / args.name + arm_dir.mkdir(mode=0o700) + state, probe_dir, work = arm_dir / "state", arm_dir / "probe", arm_dir / "work" + for path in (state, state / "runtime", state / "shared", probe_dir, work): + path.mkdir(mode=0o777) + os.chmod(path, 0o777) + (state / "runtime/session-liveness").symlink_to("/session-liveness") + probe_src = arm_dir / "rssprobe" + probe_src.mkdir(mode=0o755) + (probe_src / "sitecustomize.py").write_text(PROBE) + os.chmod(probe_src / "sitecustomize.py", 0o644) + env = dict(env) + env.update({ + "QDL_STABLE_REDIS_URL": f"redis://{PREFIX}quota:6379/0", + "QDL_KN3_MARKET_CACHE_URL": f"redis://{PREFIX}cache:6379/0", + "QDL_STABLE_INSTANCE_ID": f"rss-query-{args.name}", + "QDL_STABLE_CONSUMER_GROUP": f"rss-query-{args.name}", + "QDL_STABLE_AUDIT_PATH": "/var/lib/qdl-kn/runtime/rss-query-audit.jsonl", + "PYTHONPATH": "/rssprobe:" + env.get("PYTHONPATH", "/app"), + "QDL_RSS_TRACEMALLOC": str(args.tracemalloc), + }) + for item in args.env: + key, _, value = item.partition("=") + env[key] = value + env_file = arm_dir / ".env" + fd = os.open(env_file, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "w") as handle: + for key, value in sorted(env.items()): + if "\n" in value: + raise RuntimeError(f"env {key} spans lines") + handle.write(f"{key}={value}\n") + query = PREFIX + "query" + meta = {"arm": args.name, "image": run_image, "src": str(args.src) if args.src else None, + "tracemalloc": args.tracemalloc, "env_overrides": args.env, "bursts": args.bursts, "mix": args.mix, + "started_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())} + try: + create = ["docker", "create", "--rm", "--name", query, "--label", LABEL, "--network", NETWORK, + "--network-alias", "query_v2_1", "--user", "10001:10001", "--read-only", + "--tmpfs", "/tmp:rw,noexec,nosuid,size=67108864", "--cpus", "1.0", "--memory", "2g", + "--pids-limit", "256", "--security-opt", "no-new-privileges:true", "--cap-drop", "ALL", + "--log-opt", "max-size=50m", "--env-file", str(env_file), *ro_mounts(mounts), + "--mount", f"type=bind,src={state},dst=/var/lib/qdl-kn", + "--mount", f"type=bind,src={probe_dir},dst=/probe", + "--mount", f"type=bind,src={probe_src},dst=/rssprobe,readonly"] + if args.src: + create += ["--mount", f"type=bind,src={args.src.resolve() / 'qdl'},dst=/app/qdl,readonly"] + sh(*create, run_image, *cmd) + env_file.unlink() + sync = PREFIX + "sync" + (work / "products.json").write_text(json.dumps(products)) + os.chmod(work / "products.json", 0o644) + sh("docker", "create", "--rm", "--name", sync, "--label", LABEL, "--network", NETWORK, + "--user", "10001:10001", "--read-only", "--cpus", "0.25", "--memory", "256m", "-e", "PYTHONPATH=/app", + "-e", "SRC_URL=redis://market_cache:6379/0", "-e", f"DST_URL=redis://{PREFIX}cache:6379/0", + "-e", "SYNC_SECONDS=15", "-e", f"SYNC_DEADLINE_SECONDS={args.bursts * (args.gap_s + 90) + 900}", + *ro_mounts({"/runtime": mounts["/runtime"]}), "--mount", f"type=bind,src={work},dst=/work,readonly", + QUERY_IMAGE, "python", "-c", COPY, "/work/products.json", str(len(products)), str(COPY_ROWS)) + sh("docker", "network", "connect", PROD_INTERNAL, sync) + sh("docker", "start", sync) + sh("docker", "start", query) + time.sleep(args.startup_s) + alpha = 0 + for product in products: + if product["consumer"] == "trading-system.paper.stable" or args.mix == "ts1000": + product["rows"] = ROWS + else: + product["rows"] = STAGE_ALPHA_ROWS[alpha % len(STAGE_ALPHA_ROWS)] + alpha += 1 + meta["rows"] = [p["rows"] for p in products] + (work / "driver.json").write_text(json.dumps({ + "url": "https://query_v2_1:8200", "products": products, "bursts": args.bursts, + "rounds": 4, "settle_s": 5, "gap_s": args.gap_s, "tracemalloc": bool(args.tracemalloc)})) + os.chmod(work / "driver.json", 0o644) + identities = [a for slug in ("trading-system", "alpha-binance", "alpha-okx") for kind in ("", "-jwt") + for a in ("--mount", f"type=bind,src={IDENTITIES / (slug + kind)},dst=/identities/{slug + kind},readonly")] + driver = subprocess.run( + ["docker", "run", "--rm", "--name", PREFIX + "driver", "--label", LABEL, "--network", NETWORK, + "--user", "10001:10001", "--read-only", "--cpus", "1.0", "--memory", "1g", "-e", "PYTHONPATH=/app", + *identities, "--mount", f"type=bind,src={work},dst=/work", + "--mount", f"type=bind,src={probe_dir},dst=/probe", QUERY_IMAGE, "python", "-c", DRIVER], + capture_output=True, text=True, timeout=args.bursts * (args.gap_s + 120) + 600) + (arm_dir / "driver.out").write_text(driver.stdout + "\n--stderr--\n" + driver.stderr[-20000:]) + meta["driver_returncode"] = driver.returncode + print(driver.stdout[-6000:]) + if driver.returncode != 0: + print(driver.stderr[-3000:], file=sys.stderr) + finally: + logs = subprocess.run(["docker", "logs", "-t", query], capture_output=True, text=True, timeout=120) + (arm_dir / "query.log").write_text(logs.stdout + logs.stderr) + sh("docker", "rm", "-f", query, check=False) + sh("docker", "rm", "-f", PREFIX + "driver", check=False) + sync_logs = subprocess.run(["docker", "logs", PREFIX + "sync"], capture_output=True, text=True, timeout=60) + (arm_dir / "sync.log").write_text(sync_logs.stdout[-4000:] + sync_logs.stderr[-4000:]) + sh("docker", "rm", "-f", PREFIX + "sync", check=False) + env_file.unlink(missing_ok=True) + meta["ended_utc"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + (arm_dir / "run.json").write_text(json.dumps(meta, indent=1) + "\n") + return meta.get("driver_returncode", 1) + + +def cleanup(_args) -> int: + names = [n for n in sh("docker", "ps", "-a", "--filter", f"label={LABEL}", "--format", "{{.Names}}", + check=False).splitlines() if n.startswith(PREFIX)] + for name in names: + sh("docker", "rm", "-f", name, check=False) + nets = [n for n in sh("docker", "network", "ls", "--filter", f"label={LABEL}", "--format", "{{.Name}}", + check=False).splitlines() if n == NETWORK] + for net in nets: + sh("docker", "network", "rm", net, check=False) + print(json.dumps({"removed_containers": names, "removed_networks": nets})) + return 0 + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + sub = parser.add_subparsers(dest="command", required=True) + p = sub.add_parser("setup") + p.add_argument("--output", type=Path, required=True) + p.add_argument("--slices", type=int, default=SLICES) + p.add_argument("--lock-owner", default="claude-idr129-query-rss") + p.set_defaults(func=setup) + p = sub.add_parser("arm") + p.add_argument("--output", type=Path, required=True) + p.add_argument("--name", required=True) + p.add_argument("--src", type=Path) + p.add_argument("--image") + p.add_argument("--bursts", type=int, default=12) + p.add_argument("--gap-s", type=int, default=10) + p.add_argument("--startup-s", type=int, default=25) + p.add_argument("--tracemalloc", type=int, default=0) + p.add_argument("--mix", choices=("stage", "ts1000"), default="stage") + p.add_argument("--env", action="append", default=[]) + p.add_argument("--lock-owner", default="claude-idr129-query-rss") + p.set_defaults(func=arm) + p = sub.add_parser("cleanup") + p.set_defaults(func=cleanup) + args = parser.parse_args() + return args.func(args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_query_idr129_memory.py b/tests/test_query_idr129_memory.py new file mode 100644 index 00000000..a2430055 --- /dev/null +++ b/tests/test_query_idr129_memory.py @@ -0,0 +1,360 @@ +"""IDR-129: Query memory kept after warmup bursts. + +Measured 2026-10-19 on an isolated replica of the production Query image under repeated 64-warmup bursts +(``scripts/idr129_query_rss_isolated.py``): RSS 472 -> 683 MiB over 12 bursts. Two retained pieces: + +* A large warmup body went to the server as one ``http.response.body`` message. asyncio's TLS transport + encrypts each ``transport.write`` into the connection's OpenSSL output ``ssl.MemoryBIO``, which keeps the + capacity of its largest write for the connection's life, and a closed connection's objects are cyclic + garbage that only a rare full collection frees (130 MiB of malloc memory freed from 39 closed + connections). ``SlicedBodyResponse`` hands the same bytes over in bounded slices. +* A refused batch item's error kept its traceback, whose frames hold the error again (the IDR-114 residual + ``work``/executor cycle). The batch keeps the typed problem and releases the frames. +""" + +from __future__ import annotations + +import asyncio +import gc +import importlib +import ssl +from types import SimpleNamespace +import unittest + +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from qdl.query import ( + AccessPurpose, + BatchRequirement, + CanonicalErrorCode, + ConsumerGrade, + DataRequirement, + FeedType, + QueryProblem, + V2QueryService, +) +from qdl.query.service import QueryServiceError, _LocalBatchAdmission, _queued_local_batch_lane_policy +from qdl.runtime.memory_maintenance import ( + DEFAULT_SOFT_LIMIT_MIB, + SOFT_LIMIT_ENV, + MemoryMaintenance, + install_query_memory_maintenance, + soft_limit_from_environment, +) +from qdl.warmup import WarmupSpecification +from qdl.warmup.executor import BoundedWarmupExecutor + +router_module = importlib.import_module("qdl.api_v2.router") +SlicedBodyResponse = router_module.SlicedBodyResponse +SLICE = router_module._RESPONSE_SLICE_BYTES + + +async def _messages(response) -> list[dict]: + sent: list[dict] = [] + + async def send(message): + sent.append(message) + + async def receive(): # pragma: no cover - a response never reads + return {"type": "http.disconnect"} + + await response({"type": "http", "method": "GET"}, receive, send) + return sent + + +class SlicedBodyResponseTests(unittest.IsolatedAsyncioTestCase): + async def test_large_body_is_sent_in_bounded_slices_with_the_same_bytes_and_length(self): + body = bytes(range(256)) * (SLICE * 3 // 256 + 17) # 3 slices and a partial one + response = SlicedBodyResponse(content=body, media_type="application/json") + messages = await _messages(response) + start, *chunks = messages + self.assertEqual(start["type"], "http.response.start") + self.assertEqual(start["status"], 200) + headers = dict(start["headers"]) + self.assertEqual(headers[b"content-length"], str(len(body)).encode()) + self.assertEqual(headers[b"content-type"], b"application/json") + self.assertEqual(len(chunks), 4) + self.assertTrue(all(len(chunk["body"]) <= SLICE for chunk in chunks)) + self.assertTrue(all(type(chunk["body"]) is bytes for chunk in chunks)) + self.assertEqual([chunk["more_body"] for chunk in chunks], [True, True, True, False]) + self.assertEqual(b"".join(chunk["body"] for chunk in chunks), body) + + async def test_small_and_empty_bodies_are_one_message_as_before(self): + for body in (b"", b"{}", b"x" * SLICE): + messages = await _messages(SlicedBodyResponse(content=body, media_type="application/json")) + self.assertEqual(len(messages), 2) + self.assertEqual(messages[1], {"type": "http.response.body", "body": body}) + + async def test_warmup_responses_use_the_sliced_response(self): + body = b'{"data":[' + b"1," * SLICE + b'1]}' + original = router_module._render_warmup_result + router_module._render_warmup_result = lambda result: body + try: + response = await router_module._warmup_json_off_loop(lambda: None) + finally: + router_module._render_warmup_result = original + self.assertIsInstance(response, SlicedBodyResponse) + self.assertEqual(response.body, body) + + +class SlicedBodyOverHttpTests(unittest.TestCase): + def test_client_receives_identical_bytes_and_content_length(self): + body = b"a" * (SLICE * 2 + 5) + app = FastAPI() + + @app.get("/body") + async def endpoint(): + return SlicedBodyResponse(content=body, media_type="application/json") + + with TestClient(app) as client: + response = client.get("/body") + self.assertEqual(response.status_code, 200) + self.assertEqual(response.content, body) + self.assertEqual(response.headers["content-length"], str(len(body))) + self.assertNotIn("transfer-encoding", response.headers) + + +class TlsOutputBufferTests(unittest.TestCase): + """The stdlib/OpenSSL property the slices rely on, checked through ``ssl.MemoryBIO`` byte counts. + + asyncio's ``sslproto`` writes a ``transport.write`` into the SSL object at once and then drains the + output BIO; the encrypted bytes of one write all pass through the BIO before the first is read out. + """ + + def test_one_write_puts_the_whole_body_in_the_output_bio_and_slices_bound_it(self): + server_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + client_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + client_ctx.check_hostname = False + client_ctx.verify_mode = ssl.CERT_NONE + cert = _self_signed_pair() + if cert is None: + self.skipTest("no certificate tooling (cryptography) in this environment") + server_ctx.load_cert_chain(*cert) + + def connected(): + si, so, ci, co = ssl.MemoryBIO(), ssl.MemoryBIO(), ssl.MemoryBIO(), ssl.MemoryBIO() + server = server_ctx.wrap_bio(si, so, server_side=True) + client = client_ctx.wrap_bio(ci, co, server_side=False) + for _ in range(10): + for side in (client, server): + try: + side.do_handshake() + except ssl.SSLWantReadError: + pass + ci.write(so.read()) + si.write(co.read()) + return server, so + + body = b"x" * (4 * SLICE) + server, out = connected() + server.write(body) # one transport.write of the whole body + self.assertGreater(out.pending, len(body)) + out.read() + server, out = connected() + peak = 0 + for start in range(0, len(body), SLICE): # SlicedBodyResponse: one write per slice + server.write(body[start:start + SLICE]) + peak = max(peak, out.pending) + out.read() + self.assertLess(peak, SLICE + 16 * 1024) + + +def _self_signed_pair(): + try: + from cryptography import x509 + from cryptography.hazmat.primitives import hashes, serialization + from cryptography.hazmat.primitives.asymmetric import ec + from cryptography.x509.oid import NameOID + except ImportError: + return None + import datetime + import os + import tempfile + + key = ec.generate_private_key(ec.SECP256R1()) + name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "localhost")]) + now = datetime.datetime.now(datetime.timezone.utc) + cert = (x509.CertificateBuilder().subject_name(name).issuer_name(name).public_key(key.public_key()) + .serial_number(x509.random_serial_number()).not_valid_before(now - datetime.timedelta(minutes=1)) + .not_valid_after(now + datetime.timedelta(hours=1)).sign(key, hashes.SHA256())) + directory = tempfile.mkdtemp(prefix="idr129-tls-") + cert_path, key_path = os.path.join(directory, "c.pem"), os.path.join(directory, "k.pem") + with open(cert_path, "wb") as handle: + handle.write(cert.public_bytes(serialization.Encoding.PEM)) + with open(key_path, "wb") as handle: + handle.write(key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, + serialization.NoEncryption())) + return cert_path, key_path + + +def _requirement(uid: str, rows: int) -> DataRequirement: + return DataRequirement( + instrument_uid=uid, feed=FeedType.BAR, consumer_grade=ConsumerGrade.ALPHA, + source_policy_id="crypto_primary_v2", interval="1m", + warmup=WarmupSpecification.for_rows(rows, deadline_ms=5_000), + ) + + +def _service_refusing_items(): + """The production shape of a refused local item: history_many hands back a typed error per item, + ``work`` raises it and the real bounded executor stores it on the item's execution.""" + + class Backend: + @staticmethod + def warmup_is_local(_requirement): + return True + + @staticmethod + def history_many(requirements): + return { + requirement: QueryServiceError( + QueryProblem(CanonicalErrorCode.DATA_STALE, "stale", True), + request_id="req-1", instrument_uid=requirement.instrument_uid, + ) + for requirement in requirements + } + + class Service(V2QueryService): + def __init__(self): + self.backend = Backend() + self.instruments = SimpleNamespace() + self.warmup_executor = BoundedWarmupExecutor() + self.last_batch_evidence = {} + self._small_local_warmup_lane = True + self._queued_local_batch_lane = True + self._local_batch_admission = _LocalBatchAdmission(policy=_queued_local_batch_lane_policy()) + + return Service() + + +class RefusedItemCycleTests(unittest.TestCase): + def test_a_refused_item_keeps_its_problem_and_leaves_no_work_frame_cycle(self): + service = _service_refusing_items() + + async def warm(): + async def completion(result): + return result + return await service.warmup_batch_completed_async( + BatchRequirement(consumer_id="alpha.okx", requirements=(_requirement("a", 2500), + _requirement("b", 2500))), + purpose=AccessPurpose.INTERNAL_ALPHA, completion=completion, + ) + + loop = asyncio.new_event_loop() + try: + loop.run_until_complete(warm()) # warm imports and lazily built pools + gc.collect() + gc.set_debug(gc.DEBUG_SAVEALL) + try: + result = loop.run_until_complete(warm()) + del result + gc.collect() + finally: + gc.set_debug(0) + frames = [ + f"{item.f_code.co_filename}:{item.f_code.co_qualname}" + for item in gc.garbage if type(item).__name__ == "frame" + ] + gc.garbage.clear() + result = loop.run_until_complete(warm()) + finally: + loop.close() + self.assertEqual([item.status for item in result.results], ["DATA_STALE", "DATA_STALE"]) + self.assertTrue(all(item.problem.retryable for item in result.results)) + self.assertEqual([f for f in frames if "qdl/" in f], []) + + +class _Clock: + def __init__(self): + self.now = 1000.0 + + def __call__(self): + return self.now + + +class MemoryMaintenanceTests(unittest.TestCase): + def _maintenance(self, *, rss=900.0, busy=False, **kwargs): + self.clock, self.calls, self.records = _Clock(), [], [] + self.state = {"rss": rss, "busy": busy} + return MemoryMaintenance( + busy=lambda: self.state["busy"], rss=lambda: self.state["rss"], + collect=lambda: self.calls.append("collect") or 7, trim=lambda: self.calls.append("trim") or 1, + clock=self.clock, wall_ms=lambda: 1, record=self.records.append, **kwargs, + ) + + def test_runs_one_collection_and_trim_when_idle_and_above_the_soft_limit(self): + maintenance = self._maintenance(soft_limit_mib=512, idle_seconds=3, min_interval_seconds=120) + self.assertIsNone(maintenance.tick()) # idle for 0 s since construction + self.clock.now += 3 + record = maintenance.tick() + self.assertEqual(self.calls, ["collect", "trim"]) + self.assertEqual(record["kind"], "memory_maintenance") + self.assertEqual((record["collected"], record["soft_limit_mib"], record["rss_before_mib"]), (7, 512, 900.0)) + self.assertEqual(self.records, [record]) + + def test_never_runs_below_the_soft_limit_while_busy_or_inside_the_interval(self): + maintenance = self._maintenance(soft_limit_mib=512, idle_seconds=3, min_interval_seconds=120) + self.state["rss"] = 400.0 + self.clock.now += 10 + self.assertIsNone(maintenance.tick()) + self.state.update(rss=900.0, busy=True) + self.assertIsNone(maintenance.tick()) + self.state["busy"] = False + self.clock.now += 2 # the lane was busy 2 s ago + self.assertIsNone(maintenance.tick()) + self.clock.now += 1 + self.assertIsNotNone(maintenance.tick()) + self.clock.now += 119 + self.assertIsNone(maintenance.tick()) + self.clock.now += 1 + self.assertIsNotNone(maintenance.tick()) + self.assertEqual(self.calls, ["collect", "trim"] * 2) + + def test_soft_limit_zero_disables_and_the_environment_is_validated(self): + maintenance = self._maintenance(soft_limit_mib=0, idle_seconds=0, min_interval_seconds=0) + self.clock.now += 100 + self.assertIsNone(maintenance.tick()) + self.assertEqual(self.calls, []) + self.assertEqual(soft_limit_from_environment({}), DEFAULT_SOFT_LIMIT_MIB) + self.assertEqual(soft_limit_from_environment({SOFT_LIMIT_ENV: "0"}), 0) + self.assertEqual(soft_limit_from_environment({SOFT_LIMIT_ENV: "1024"}), 1024) + with self.assertRaises(ValueError): + soft_limit_from_environment({SOFT_LIMIT_ENV: "-1"}) + + def test_a_failing_recorder_does_not_break_maintenance(self): + maintenance = MemoryMaintenance(busy=lambda: False, rss=lambda: 900.0, collect=lambda: 0, trim=lambda: 0, + idle_seconds=0, min_interval_seconds=0, record=_raise) + self.assertIsNotNone(maintenance.tick()) + + def test_installed_on_the_app_and_runs_with_its_lifespan(self): + app = FastAPI() + service = SimpleNamespace(large_warmup_lane_busy=lambda: True) + maintenance = install_query_memory_maintenance(app, service) + self.assertIs(app.state.memory_maintenance, maintenance) + with TestClient(app): + pass + self.assertEqual(maintenance.runs, 0) # the lane is busy + + def test_the_lane_busy_signal_follows_the_large_warmup_lane(self): + service = _service_refusing_items() + self.assertFalse(service.large_warmup_lane_busy()) + + async def hold(): + gate = asyncio.Event() + task = asyncio.create_task(service._local_batch_admission.run(gate.wait, consumer_id="a")) + await asyncio.sleep(0) + busy = service.large_warmup_lane_busy() + gate.set() + await task + return busy, service.large_warmup_lane_busy() + + self.assertEqual(asyncio.run(hold()), (True, False)) + + +def _raise(_record): + raise RuntimeError("recorder down") + + +if __name__ == "__main__": # pragma: no cover + unittest.main()