fix: bound the decoder adapter caches (#191)

* fix: bound the decoder adapter caches

Each decoder cache now holds at most 4096 models and is emptied when an
insert would go past that, so models created per request no longer pile
up for the life of the decoder.

* docs: middleware response streams must match the client

* test: check only our cache for dropped msgspec models

Older msgspec builds typing.Tuple[...] while inspecting a model, and
typing's subscription LRU keeps the recent models alive, so the weakref
check failed on the dependency floors.

Author: lesnik512

docs/decoders.md

@@ -35,9 +35,13 @@ Both clients take `decoders: Sequence[ResponseDecoder] | None = None`. The list
 - If `response_model=` is set and no decoder accepts it, the client raises `MissingDecoderError` before sending the request. That means nothing handles the type, and the fix is to install an extra or pass `decoders=[...]`. A `DecodeError` instead means a decoder ran and the body didn't fit the type, which points at the server or the model. See [Errors](errors.md).
 - `MsgspecDecoder` accepts a `Struct`, dataclass, `TypedDict` or `NamedTuple` only if msgspec can decode every field type itself, nested fields included. A field typed as a pydantic model or some other class would need a `dec_hook`, which `MsgspecDecoder` doesn't take, so it declines the model and you get `MissingDecoderError` before the request goes out.
 
+## Keep response models long-lived
+
+The built-in decoders build a parser once per model (a `pydantic.TypeAdapter` or a `msgspec.json.Decoder`) and cache it, so define each response model once, at module level, and reuse it. Each decoder caches up to 4,096 models. When the cache is full, the next new model empties it. A model created per request, for example with `pydantic.create_model` or `msgspec.defstruct`, takes a new slot every time, so the cache keeps filling up and emptying, and every model seen after it empties gets its parser built again.
+
 ## One sync protocol for both clients
 
-Middleware comes in sync and async versions, but there is only one `ResponseDecoder` protocol, used by both clients. `decode` is synchronous because the client has the whole body by the time it runs, so there is nothing to await. If a middleware returns a streamed response it never read, the client reads it before decoding. In `AsyncClient` that stream also has to support sync iteration, as `httpx2.ByteStream` does; otherwise the middleware should `await response.aread()` before returning it. The same decoder works with either client.
+Middleware comes in sync and async versions, but there is only one `ResponseDecoder` protocol, used by both clients. `decode` is synchronous because the client has the whole body by the time it runs, so there is nothing to await. If a middleware returns a streamed response it never read, the client reads it before decoding. The stream has to match the client: async in `AsyncClient` and sync in `Client`, or the call raises `TypeError` (see [Middleware](middleware.md)). An in-memory body such as `content=b"..."` works in both. The same decoder works with either client.
 
 ## Writing your own
 

src/httpware/decoders/_caching.py

@@ -10,6 +10,8 @@
 
 V = typing.TypeVar("V")
 
+_MAX_ENTRIES = 4096
+
 _Cache: typing.TypeAlias = dict[Hashable, tuple[object, V | None]]
 
 
@@ -28,14 +30,15 @@ def _lookup(cache: "_Cache[V]", model: object, build: Callable[[typing.Any], V |
     """Return the cached `build(model)`, building it on a miss; None means `model` cannot be decoded.
 
     `build` must not raise. Concurrent first calls may build twice; the last write wins.
+    A full cache is emptied before the next insert, so a model seen again after that is rebuilt.
     """
     try:
         key = _key(model)
         entry = cache.get(key)
     except Exception:  # noqa: BLE001 — unhashable models fall back to a repr key
         return _lookup_by_repr(cache, model, build)
     if entry is None:
-        entry = cache[key] = (model, build(model))
+        entry = _store(cache, key, model, build)
     return entry[1]
 
 
@@ -48,5 +51,15 @@ def _lookup_by_repr(cache: "_Cache[V]", model: object, build: Callable[[typing.A
             return entry[1]
     except Exception:  # noqa: BLE001 — a model that cannot be keyed is built uncached
         return build(model)
-    entry = cache[key] = (model, build(model))
-    return entry[1]
+    return _store(cache, key, model, build)[1]
+
+
+def _store(
+    cache: "_Cache[V]", key: Hashable, model: object, build: Callable[[typing.Any], V | None]
+) -> tuple[object, V | None]:
+    """Build and cache the entry for `model`, emptying `cache` first once it holds `_MAX_ENTRIES`."""
+    entry = (model, build(model))
+    if len(cache) >= _MAX_ENTRIES:
+        cache.clear()
+    cache[key] = entry
+    return entry

tests/test_decoders_caching.py

@@ -0,0 +1,77 @@
+"""The decoder adapter caches stay bounded when models are created per request."""
+
+import concurrent.futures
+import gc
+import typing
+import weakref
+
+import msgspec
+import pydantic
+import pytest
+
+from httpware.decoders import _caching
+from httpware.decoders.msgspec import MsgspecDecoder
+from httpware.decoders.pydantic import PydanticDecoder
+
+
+_CAP = 8
+
+
+@pytest.fixture(autouse=True)
+def _small_cap(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(_caching, "_MAX_ENTRIES", _CAP)
+
+
+def _decode_through_fresh_models(decoder: PydanticDecoder | MsgspecDecoder, models: typing.Iterator[type]) -> list:
+    refs = []
+    for model in models:
+        decoder.decode(b'{"x": 1}', model)
+        refs.append(weakref.ref(model))
+    gc.collect()
+    return refs
+
+
+def test_pydantic_cache_is_bounded_and_releases_old_models() -> None:
+    decoder = PydanticDecoder()
+    models = (pydantic.create_model(f"M{i}", x=(int, ...)) for i in range(_CAP + 1))
+    refs = _decode_through_fresh_models(decoder, models)
+    assert len(decoder._adapters) <= _CAP  # noqa: SLF001
+    assert refs[0]() is None
+
+
+def test_msgspec_cache_is_bounded_and_drops_old_models() -> None:
+    """Checks our cache only: older msgspec inspects through `typing.Tuple[...]`, whose LRU keeps recent models."""
+    decoder = MsgspecDecoder()
+    models = [msgspec.defstruct(f"S{i}", [("x", int)]) for i in range(_CAP + 1)]
+    for model in models:
+        decoder.decode(b'{"x": 1}', model)
+    assert len(decoder._adapters) <= _CAP  # noqa: SLF001
+    assert all(cached is not models[0] for cached, _ in decoder._adapters.values())  # noqa: SLF001
+
+
+def test_repr_keyed_cache_is_bounded() -> None:
+    decoder = PydanticDecoder()
+    for i in range(_CAP + 1):
+        decoder.decode(b"1", typing.Annotated[int, [i]])  # ty: ignore[invalid-argument-type]
+    assert len(decoder._adapters) <= _CAP  # noqa: SLF001
+
+
+def test_model_decodes_after_the_cache_is_cleared() -> None:
+    decoder = PydanticDecoder()
+    for i in range(_CAP + 1):
+        decoder.decode(b'{"x": 1}', pydantic.create_model(f"M{i}", x=(int, ...)))
+    model = pydantic.create_model("Late", x=(int, ...))
+    assert decoder.decode(b'{"x": 2}', model).model_dump() == {"x": 2}
+    assert decoder.can_decode(model)
+
+
+@pytest.mark.stress
+def test_clear_racing_with_lookups_returns_the_right_adapter() -> None:
+    models = [type(f"T{i}", (), {}) for i in range(_CAP * 4)]
+    cache: _caching._Cache[type] = {}
+
+    def look_up_all(_: int) -> bool:
+        return all(_caching._lookup(cache, model, lambda m: m) is model for _ in range(50) for model in models)  # noqa: SLF001
+
+    with concurrent.futures.ThreadPoolExecutor(max_workers=16) as pool:
+        assert all(pool.map(look_up_all, range(32)))