diff --git a/gmlx/cache/kvarn_apc.py b/gmlx/cache/kvarn_apc.py index d54a3474..39454ad0 100644 --- a/gmlx/cache/kvarn_apc.py +++ b/gmlx/cache/kvarn_apc.py @@ -31,6 +31,10 @@ import hashlib import importlib +import logging +import secrets + +_log = logging.getLogger(__name__) _FLAG = "_gmlx_kvarn_apc" _MODE_STAMP = "_gmlx_kvarn_apc_exact" @@ -99,15 +103,31 @@ def apply_kvarn_salt(manager, model) -> None: pairing, gated on the model actually converting: a kvarn-window boot of a zero-conversion arch (deepseek4, recurrent_gemma) runs pure fp16 caches, and salting its entries would cold-miss every cross-boot - lookup. Failures leave the salt at its XOR-identity default.""" + lookup. The XOR-identity default is correct only for that + non-converting case; a converting model whose salt computation fails + gets a random per-boot salt instead, so its entries stay out of the + fp16 hash space and cold-miss across boots rather than warm-adopt + under the wrong wire config.""" + if manager is None or model is None: + return + if not kvarn_model_converts(model): + return try: - if manager is None or model is None: - return salt = kvarn_entry_salt(model) - if salt and kvarn_model_converts(model): - manager._exact_extra_salt = salt except Exception: - pass + salt = secrets.randbits(63) | 1 + _log.warning( + "kvarn exact-tier salt computation failed; salting with a " + "per-boot value (cross-boot APC reuse off for this model)", + exc_info=True) + if not salt: + return + try: + manager._exact_extra_salt = salt + except Exception: + _log.warning( + "kvarn exact-tier salt could not be applied; kvarn entries " + "share the fp16 hash space until the next boot", exc_info=True) def kvarn_entry_salt(model=None) -> int: diff --git a/gmlx/cache/snapshot.py b/gmlx/cache/snapshot.py index c21437df..5dc4754b 100644 --- a/gmlx/cache/snapshot.py +++ b/gmlx/cache/snapshot.py @@ -1571,12 +1571,17 @@ def _once(): len(states), rec.nbytes / (1 << 20)) return p except Exception: + # One swallow per call: a failing decline (telemetry) or first + # release must not leak the other block set out of the manager. try: _ckpt_decline(manager, "exception") - manager.release(main_blocks) - manager.release(bounded_blocks) except Exception: - pass # best-effort release on the failure path + pass # decline is telemetry only + for blks in (main_blocks, bounded_blocks): + try: + manager.release(blks) + except Exception: + pass # best-effort release on the failure path _log.warning("APC ckpt store failed; continuing", exc_info=True) return 0 @@ -2155,11 +2160,13 @@ def _ckpt_disk_lookup(manager, ids, *, extra_hash, min_prefix_tokens, _log.info("APC ckpt hit: prefix=%d (disk skeleton)", p) return warm, p except Exception: - try: - manager.release(blocks) - manager.release(wblocks) - except Exception: - pass # best-effort release on the failure path + # One swallow per release: a failing first release must not leak + # the second block set out of the manager. + for blks in (blocks, wblocks): + try: + manager.release(blks) + except Exception: + pass # best-effort release on the failure path _log.warning("APC ckpt disk lookup failed; continuing", exc_info=True) return None, 0 diff --git a/gmlx/models/gemma4/batched_sdpa.py b/gmlx/models/gemma4/batched_sdpa.py index 9f5981c7..9f532a36 100644 --- a/gmlx/models/gemma4/batched_sdpa.py +++ b/gmlx/models/gemma4/batched_sdpa.py @@ -222,8 +222,8 @@ def _claim(queries, keys, values, cache, scale, mask, sinks): starts=starts) _ONECALL[0] += 1 return out - except Exception: - pass # op-build rejection -> per-row loop + except Exception as exc: # op-build rejection -> per-row loop + attn_hd512._warn_fallback_once("g4_batched_decode", exc) # qL==1 needs no mask after the tail slice; verify blocks (qL 2..8) # occupy the LAST qL key positions, which is exactly mx.fast's # end-aligned "causal" semantics on the sliced row. diff --git a/gmlx/serve/governor.py b/gmlx/serve/governor.py index 3848ffa9..e7b1a007 100644 --- a/gmlx/serve/governor.py +++ b/gmlx/serve/governor.py @@ -980,6 +980,7 @@ def install_governor() -> bool: if getattr(_ar.BatchGenerator._next, _INSTALLED_FLAG, False): return True _orig = _ar.BatchGenerator._next + harvest_warned = [False] def _governed_next(self, **kwargs): try: @@ -991,7 +992,10 @@ def _governed_next(self, **kwargs): try: _harvest_tick(self, _state(self), out) except Exception: - pass + if not harvest_warned[0]: + harvest_warned[0] = True + _log.warning("[governor] harvest failed; band inputs may " + "go stale (warn-once)", exc_info=True) return out setattr(_governed_next, _INSTALLED_FLAG, True) diff --git a/gmlx/serve/patches/completions.py b/gmlx/serve/patches/completions.py index b8fc08af..e2d6b250 100644 --- a/gmlx/serve/patches/completions.py +++ b/gmlx/serve/patches/completions.py @@ -15,6 +15,7 @@ import importlib import inspect import json +import logging import time import uuid from typing import Any, List, Optional @@ -29,6 +30,8 @@ _COMPLETIONS_PATHS = ("/completions", "/v1/completions") +_log = logging.getLogger(__name__) + # Bound once at install time (the route cannot run before install); the # sibling patches bind their upstream modules the same way. _app_mod = None @@ -126,13 +129,20 @@ def _include_usage(request: CompletionRequest) -> bool: return bool(getattr(so, "include_usage", False)) +_RECORD_FAILURE_WARNED = False + + def _record_failure(runtime, model: str, stream: bool, error: str) -> None: + global _RECORD_FAILURE_WARNED try: runtime.metrics.record_failure(endpoint="/v1/completions", model=model, stream=stream, error=error) except Exception: - pass + if not _RECORD_FAILURE_WARNED: + _RECORD_FAILURE_WARNED = True + _log.warning("failure-metrics record raised; failed requests " + "are missing from /v1/metrics", exc_info=True) def _completion_envelope(gen_mod, *, model, stream, prompt_tokens, diff --git a/gmlx/serve/patches/request_flow.py b/gmlx/serve/patches/request_flow.py index cc61dcf5..5a2f1339 100644 --- a/gmlx/serve/patches/request_flow.py +++ b/gmlx/serve/patches/request_flow.py @@ -5,6 +5,7 @@ from __future__ import annotations import importlib +import logging import gmlx.serve.bridge_vlm as serving @@ -15,6 +16,8 @@ ) from . import routes as _routes +_log = logging.getLogger(__name__) + # Off-loop model load - keep /health (and siblings) responsive during a load _LOAD_OFFLOAD_FLAG = "_kq_gguf_load_offload" @@ -232,8 +235,11 @@ async def _pump(): task.cancel() try: await task - except BaseException: - pass + except asyncio.CancelledError: + pass # the expected unwind of our own cancel + except Exception: + _log.warning("SSE pump task died with an unreported error", + exc_info=True) aclose = getattr(body, "aclose", None) if aclose is not None: try: diff --git a/gmlx/serve/patches/routes.py b/gmlx/serve/patches/routes.py index f04039fa..5160bcaa 100644 --- a/gmlx/serve/patches/routes.py +++ b/gmlx/serve/patches/routes.py @@ -6,6 +6,7 @@ from __future__ import annotations import importlib +import logging import os import time @@ -20,6 +21,8 @@ _remove_routes, ) +_log = logging.getLogger(__name__) + def _mtime(path) -> int: try: @@ -854,9 +857,17 @@ def _release_preload_holds(pool, path, *, only_evicted: bool = False) -> int: try: if hasattr(pool, "unmark_retained"): pool.unmark_retained(hold) + except Exception: + _log.warning("preload hold unmark failed; releasing anyway", + exc_info=True) + try: hold.release() except Exception: - pass + # A dropped-but-unreleased hold would pin the model resident + # for the process lifetime; keep it tracked for a retry. + _log.warning("preload hold release failed; keeping it tracked", + exc_info=True) + continue _PRELOAD_HOLDS.remove(hold) released += 1 return released diff --git a/gmlx/serve/patches/sampling.py b/gmlx/serve/patches/sampling.py index 34e16e7c..177f34f0 100644 --- a/gmlx/serve/patches/sampling.py +++ b/gmlx/serve/patches/sampling.py @@ -4,12 +4,16 @@ from __future__ import annotations +import logging + import gmlx.serve.bridge_vlm as serving from ._common import ( _PATCH_FLAG, _install_gen_args_transform, ) +_log = logging.getLogger(__name__) + # Sampling-profile injection # GenerationArguments attr <- profile sampling key (1:1 names). Each maps to the @@ -78,9 +82,22 @@ def _effective_request_param(request, spec, key, default=None): return default +_NEWLINE_PROBE_WARNED = False + + +def _warn_newline_probe_once(exc: BaseException) -> None: + global _NEWLINE_PROBE_WARNED + if not _NEWLINE_PROBE_WARNED: + _NEWLINE_PROBE_WARNED = True + _log.warning( + "XTC newline probe failed (%s: %s); newline stays maskable " + "by XTC for this process", type(exc).__name__, exc) + + def _xtc_special_tokens(processor) -> list: """Newline + EOS token ids, excluded from XTC masking (the same convention - as the run/chat CLI). Defensive: any tokenizer shape miss degrades to [].""" + as the run/chat CLI). Defensive: any tokenizer shape miss degrades to the + ids that did resolve, with a warn-once for the newline probe.""" tok = getattr(processor, "tokenizer", processor) if tok is None: return [] @@ -90,10 +107,10 @@ def _xtc_special_tokens(processor) -> list: except TypeError: try: ids.extend(tok.encode("\n")) - except Exception: - pass - except Exception: - pass + except Exception as exc: + _warn_newline_probe_once(exc) + except Exception as exc: + _warn_newline_probe_once(exc) eos = getattr(tok, "eos_token_ids", None) if eos is None: eos = getattr(tok, "eos_token_id", None) diff --git a/gmlx/serve/residency.py b/gmlx/serve/residency.py index 8566633a..6dcf5de0 100644 --- a/gmlx/serve/residency.py +++ b/gmlx/serve/residency.py @@ -412,7 +412,8 @@ def _stamp_apc_mode(rg) -> None: try: rg.apc_mode = mode except Exception: - pass + _log.warning("apc_mode stamp failed; the generator keeps its " + "default mode", exc_info=True) class _ResidencyPool: diff --git a/gmlx/spec/engine.py b/gmlx/spec/engine.py index 88729b5b..873cd7a2 100644 --- a/gmlx/spec/engine.py +++ b/gmlx/spec/engine.py @@ -153,7 +153,10 @@ def _init_with_stash(self, model, processor, **kwargs): try: model._kq_apc_manager = kwargs.get("apc_manager") except Exception: - pass + if kwargs.get("apc_manager") is not None: + _log.warning( + "APC OFF for this model: manager stash failed -- " + "every request prefills cold", exc_info=True) _orig_init(self, model, processor, **kwargs) # Stock admission forms a prompt batch only when free slots >= # prefill_batch_size. Stock pairs 32/8 (24 slots stay open); the @@ -289,7 +292,12 @@ def _ckpt_layout_for(model, block_size: int = 16): try: tags = tuple(ckpt_layout(lm.make_cache(), block_size) or ()) except Exception: + # An empty stash reads as "no ckpt signature" below, never as + # a valid all-empty layout that could sign entries. tags = () + _log.warning( + "APC ckpt layout probe failed; ckpt tier off for this " + "model", exc_info=True) try: model._kq_apc_ckpt_layout = tags except Exception: diff --git a/gmlx/stream/installs.py b/gmlx/stream/installs.py index 44705c97..83b8269a 100644 --- a/gmlx/stream/installs.py +++ b/gmlx/stream/installs.py @@ -14,8 +14,11 @@ from __future__ import annotations +import logging import weakref +_log = logging.getLogger(__name__) + # The attributes the loader hangs a streaming install on. Each is closeable # and each holds host resources (shard fds, staging pools, mlocked ranges) # that must not outlive the model. @@ -136,6 +139,7 @@ def release(model) -> None: import gc owner = streaming_owner(model) + close_failed = False for attr in STREAM_ATTRS: helper = getattr(owner, attr, None) if helper is None: @@ -145,12 +149,18 @@ def release(model) -> None: try: close() except Exception: - pass + close_failed = True + _log.warning("streaming teardown: %s close failed; keeping " + "it tracked", attr, exc_info=True) + continue # keep the attr so the helper stays reachable try: object.__setattr__(owner, attr, None) except Exception: pass - _LIVE[:] = [(r, n) for r, n in _LIVE if r() is not model] + # A model with an un-closed helper keeps its wired-byte charge; the + # arena prune already keeps open feeders via _open. + if not close_failed: + _LIVE[:] = [(r, n) for r, n in _LIVE if r() is not model] _ARENAS[:] = [r for r in _ARENAS if _open(r())] del owner, model gc.collect() diff --git a/gmlx/upstream/attn_hd512.py b/gmlx/upstream/attn_hd512.py index ca08b1f3..3c84bb42 100644 --- a/gmlx/upstream/attn_hd512.py +++ b/gmlx/upstream/attn_hd512.py @@ -50,11 +50,27 @@ """ from __future__ import annotations +import logging import mlx.core as mx from gmlx.envflags import env_bool, env_int +_log = logging.getLogger(__name__) + +_FALLBACK_WARNED: set[str] = set() + + +def _warn_fallback_once(route: str, exc: BaseException) -> None: + """Kernel build failed at call time: the route stays eligible and + fails again every call, silently costing the fallback path. Eval-time + (lazy) errors escape this guard by design.""" + if route not in _FALLBACK_WARNED: + _FALLBACK_WARNED.add(route) + _log.warning("sdpa route %s failed (%s: %s); using the stock " + "fallback for such shapes", route, + type(exc).__name__, exc) + try: import mlx_kquant _HAS_COMPILED = hasattr(mlx_kquant, "sdpa_vector") @@ -454,8 +470,8 @@ def _wrapped_sdpa(q, k, v, *, scale=1.0, mask=None, **kw): _route("gqa_decode", q, k, mask) return mlx_kquant.sdpa_decode_gqa( q, k, v, float(scale), sinks=kw.get("sinks")) - except Exception: - pass # any unsupported shape -> stock fallback + except Exception as exc: # any unsupported shape -> stock fallback + _warn_fallback_once("gqa_decode", exc) if (_FA_DECODE and _HAS_FA_VERIFY and kw.get("sinks") is None and _fa_decode_eligible(q, k, v, mask)): try: @@ -466,8 +482,8 @@ def _wrapped_sdpa(q, k, v, *, scale=1.0, mask=None, **kw): mx.contiguous(q.reshape(B, kv, hq // kv, hd)), k, v, float(scale), 1) return out.reshape(B, hq, 1, hd) - except Exception: - pass # older kernel gate (q_len >= 2) -> stock fallback + except Exception as exc: # older kernel gate (q_len >= 2) -> stock + _warn_fallback_once("fa_decode", exc) if (_VERIFY_FA and _HAS_FA_VERIFY and kw.get("sinks") is None and _fa_verify_eligible(q, k, v, mask)): try: @@ -491,30 +507,30 @@ def _wrapped_sdpa(q, k, v, *, scale=1.0, mask=None, **kw): for i in range(n)], axis=2) return out.reshape(B, hq, qL, hd) - except Exception: - pass # any unsupported shape -> stock fallback + except Exception as exc: # any unsupported shape -> stock fallback + _warn_fallback_once("fa_verify", exc) if (_VERIFY_GEMM and kw.get("sinks") is None and _verify_gemm_eligible(q, k, v, mask)): try: _route("verify_gemm", q, k, mask) return _verify_gemm(q, k, v, float(scale), mask == "causal") - except Exception: - pass # any unsupported shape -> stock fallback + except Exception as exc: # any unsupported shape -> stock fallback + _warn_fallback_once("verify_gemm", exc) if (_HAS_COMPILED and _eligible(q, k, v, mask) and kw.get("sinks") is None): try: _route("sdpa_vector", q, k, mask) return mlx_kquant.sdpa_vector( q, k, v, float(scale), causal=(mask == "causal")) - except Exception: - pass # any unsupported shape -> stock fallback + except Exception as exc: # any unsupported shape -> stock fallback + _warn_fallback_once("sdpa_vector", exc) elif _prefill_eligible(q, k, v, mask): try: _route("chunked_prefill", q, k, mask) return _chunked_prefill(q, k, v, scale, mask, _PREFILL_TILE, sinks=kw.get("sinks")) - except Exception: - pass # any unsupported shape -> stock fallback + except Exception as exc: # any unsupported shape -> stock fallback + _warn_fallback_once("chunked_prefill", exc) _route("stock", q, k, mask) _stock_depth_warning(q, k, mask, kw.get("sinks")) return _orig_sdpa(q, k, v, scale=scale, mask=mask, **kw) diff --git a/gmlx/upstream/quantized_sdpa_fix.py b/gmlx/upstream/quantized_sdpa_fix.py index e18d715c..39a29677 100644 --- a/gmlx/upstream/quantized_sdpa_fix.py +++ b/gmlx/upstream/quantized_sdpa_fix.py @@ -54,12 +54,16 @@ from __future__ import annotations import collections +import logging import mlx.core as mx from gmlx.envflags import env_bool, env_int +_log = logging.getLogger(__name__) + _installed = False +_FALLBACK_WARNED = False _MODULES = ("mlx_lm.models.base", "mlx_vlm.models.base") @@ -196,8 +200,14 @@ def _masked_grouped_qsdpa(queries, q_keys, q_values, scale, mask, queries, kw, vw, float(scale), starts=starts, k_scales=ks, k_biases=kb, v_scales=vs, v_biases=vb) - except Exception: - pass # any unsupported layout -> stock fallback + except Exception as exc: # unsupported layout -> stock + global _FALLBACK_WARNED + if not _FALLBACK_WARNED: + _FALLBACK_WARNED = True + _log.warning( + "quantized sdpa route failed (%s: %s); using " + "the stock fallback for such shapes", + type(exc).__name__, exc) if ( isinstance(mask, mx.array) and mask.ndim == 4 diff --git a/tests/cache/test_ckpt_decode_lcp.py b/tests/cache/test_ckpt_decode_lcp.py index f444d89b..c5f88b95 100644 --- a/tests/cache/test_ckpt_decode_lcp.py +++ b/tests/cache/test_ckpt_decode_lcp.py @@ -362,3 +362,34 @@ def rec_store(manager, ids, cache, *, extra_hash=0, skeleton_disk=True, ckpt._ckpt_mid_prefill_store(batch) # boundary 32: interval ckpt._ckpt_mid_prefill_store(batch) # boundary 64: terminal assert seen == [(32, False), (64, True)] + + +def test_disk_lookup_failure_attempts_every_release(caplog): + """A failing release in the disk-lookup failure handler must not + skip the other block set: each release gets its own attempt.""" + from gmlx.cache.compat import runtime_cache_module + + KVCache = runtime_cache_module().KVCache + releases = [] + + class _Man: + block_size = 16 + + def lookup_exact_cache(self, ids, *, extra_hash, min_prefix_tokens): + return [KVCache()], 32 + + def lookup_prefix(self, ids, *, extra_hash): + raise RuntimeError("prefix lookup broke") + + def release(self, blocks): + releases.append(blocks) + if len(releases) == 1: + raise RuntimeError("first release broke") + + with caplog.at_level("WARNING", logger="gmlx.cache.snapshot"): + out = cs._ckpt_disk_lookup( + _Man(), list(range(40)), extra_hash=0, + min_prefix_tokens=1, layout=None) + assert out == (None, 0) + assert len(releases) == 2 + assert any("disk lookup failed" in r.message for r in caplog.records) diff --git a/tests/cache/test_kvarn_stamp_salt.py b/tests/cache/test_kvarn_stamp_salt.py index eadd1c57..fe7ee0ea 100644 --- a/tests/cache/test_kvarn_stamp_salt.py +++ b/tests/cache/test_kvarn_stamp_salt.py @@ -123,6 +123,28 @@ def test_salt_gated_on_conversion(kvarn_ops_ok): assert man._exact_extra_salt == 0, type(model).__name__ +def test_salt_failure_on_converting_model_salts_per_boot( + kvarn_ops_ok, monkeypatch, caplog): + # A converting model whose salt computation fails must not fall back + # to the XOR identity: identity keys would warm-adopt fp16 entries + # under the wrong wire config. It gets a random per-boot salt. + import gmlx.cache.kvarn_apc as ka + + def _boom(model=None): + raise RuntimeError("salt probe broke") + + monkeypatch.setattr(ka, "kvarn_entry_salt", _boom) + man = SimpleNamespace(_exact_extra_salt=0) + with caplog.at_level("WARNING", logger="gmlx.cache.kvarn_apc"): + apply_kvarn_salt(man, _stamped(_hybrid(), "kvarn")) + assert man._exact_extra_salt != 0 + assert any("per-boot" in r.message for r in caplog.records) + # Non-converting models still take the identity default on failure. + man2 = SimpleNamespace(_exact_extra_salt=0) + apply_kvarn_salt(man2, _stamped(_rec_gemma(), "kvarn")) + assert man2._exact_extra_salt == 0 + + def test_salt_zero_without_a_kvarn_stamp(kvarn_ops_ok, monkeypatch): monkeypatch.setenv("KV_QUANT_SCHEME", "kvarn") # the env is not read man = SimpleNamespace(_exact_extra_salt=0) diff --git a/tests/serve/test_patches_routes.py b/tests/serve/test_patches_routes.py index bbdea379..c0cb93e0 100644 --- a/tests/serve/test_patches_routes.py +++ b/tests/serve/test_patches_routes.py @@ -342,3 +342,35 @@ def test_keep_missing_model_field(): r = client.post("/v1/keep", json={}) assert r.status_code == 400, r.text assert r.json() == {"status": "error", "message": "missing 'model'"} + + +def test_release_preload_holds_keeps_hold_on_release_failure(caplog): + """A hold whose release() raises must stay tracked: dropping it + un-released would pin the model resident for the process lifetime.""" + from types import SimpleNamespace + + class _BadHold: + _entry = SimpleNamespace(model_path="/abs/qwen.gguf") + released = False + + def release(self): + raise RuntimeError("release broke") + + class _GoodHold: + _entry = SimpleNamespace(model_path="/abs/qwen.gguf") + released = False + + def release(self): + self.released = True + + bad, good = _BadHold(), _GoodHold() + sp_routes._PRELOAD_HOLDS[:] = [bad, good] + try: + with caplog.at_level("WARNING", logger="gmlx.serve.patches.routes"): + n = sp_routes._release_preload_holds(object(), None) + assert n == 1 + assert good.released + assert sp_routes._PRELOAD_HOLDS == [bad] + assert any("keeping it tracked" in r.message for r in caplog.records) + finally: + sp_routes._PRELOAD_HOLDS.clear() diff --git a/tests/serve/test_patches_sampling.py b/tests/serve/test_patches_sampling.py index dc82c54c..a064b199 100644 --- a/tests/serve/test_patches_sampling.py +++ b/tests/serve/test_patches_sampling.py @@ -387,6 +387,22 @@ class _IntEosTok(_Tok): types.SimpleNamespace(tokenizer=_IntEosTok())) == [7, 9] +def test_xtc_newline_probe_failure_warns_once_keeps_eos(monkeypatch, caplog): + class _BadEncodeTok: + eos_token_id = 7 + + def encode(self, s, add_special_tokens=True): + raise RuntimeError("tokenizer shape miss") + + monkeypatch.setattr(sp_sampling, "_NEWLINE_PROBE_WARNED", False) + proc = types.SimpleNamespace(tokenizer=_BadEncodeTok()) + with caplog.at_level("WARNING", logger="gmlx.serve.patches.sampling"): + assert sp_sampling._xtc_special_tokens(proc) == [7] # EOS survives + assert sp_sampling._xtc_special_tokens(proc) == [7] + warns = [r for r in caplog.records if "XTC newline probe" in r.message] + assert len(warns) == 1 + + def test_install_xtc_wraps_and_stacks_with_profile_injection(): sp.install_gen_args_profile_injection() sp.install_xtc_sampling() diff --git a/tests/spec/test_ckpt_layout_probe.py b/tests/spec/test_ckpt_layout_probe.py new file mode 100644 index 00000000..d4fb9d03 --- /dev/null +++ b/tests/spec/test_ckpt_layout_probe.py @@ -0,0 +1,23 @@ +"""_ckpt_layout_for failure handling: a broken probe reads as "no ckpt +signature", never as a valid empty layout, and it says so in the log.""" + +from __future__ import annotations + +from types import SimpleNamespace + +from gmlx.spec import engine + + +class _Broken: + def make_cache(self): + raise RuntimeError("cache probe broke") + + +def test_probe_failure_returns_none_and_warns(caplog): + model = SimpleNamespace(language_model=_Broken()) + with caplog.at_level("WARNING", logger="gmlx.spec.engine"): + assert engine._ckpt_layout_for(model) is None + assert any("ckpt layout probe failed" in r.message for r in caplog.records) + # The failure is stashed; the re-read stays None without re-probing. + model.language_model = None + assert engine._ckpt_layout_for(model) is None diff --git a/tests/stream/test_installs.py b/tests/stream/test_installs.py index 8b6cd802..13d1897f 100644 --- a/tests/stream/test_installs.py +++ b/tests/stream/test_installs.py @@ -87,17 +87,25 @@ def test_release_closes_every_helper_and_drops_the_record(): assert installs.live_wired_bytes() == 0 -def test_release_survives_a_helper_that_raises(): - # One failing close must not leak the others' arenas and fds. +def test_release_survives_a_helper_that_raises(caplog): + # One failing close must not leak the others' arenas and fds. The + # failed helper stays attached and the model keeps its wired charge: + # its mlocked ranges may still be live. class _Bad(_Helper): def close(self): raise RuntimeError("closed twice") m = _Holder() - pin = _Helper() - m._kq_decode_feeder, m._kq_weights_pin = _Bad(), pin - installs.release(m) + bad, pin = _Bad(), _Helper() + m._kq_decode_feeder, m._kq_weights_pin = bad, pin + installs.record(m, 90) + with caplog.at_level("WARNING", logger="gmlx.stream.installs"): + installs.release(m) assert pin.closed + assert m._kq_decode_feeder is bad and m._kq_weights_pin is None + assert installs.live_wired_bytes() == 90 + assert any("close failed" in r.message for r in caplog.records) + installs._LIVE.clear() def test_streaming_owner_descends_the_wrapper_chain():