fix(query): IDR-129 Query RSS kept after warmup bursts

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.

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:

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,

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