From cf1681767bfb7576fdf0740600cc9719634ee340 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sat, 17 Oct 2026 14:49:20 +0300 Subject: [PATCH 1/3] 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/decoders.md | 4 ++ src/httpware/decoders/_caching.py | 19 ++++++-- tests/test_decoders_caching.py | 75 +++++++++++++++++++++++++++++++ 3 files changed, 95 insertions(+), 3 deletions(-) create mode 100644 tests/test_decoders_caching.py diff --git a/docs/decoders.md b/docs/decoders.md index 83585a0..831633e 100644 --- a/docs/decoders.md +++ b/docs/decoders.md @@ -35,6 +35,10 @@ 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. diff --git a/src/httpware/decoders/_caching.py b/src/httpware/decoders/_caching.py index d0524ba..1d9c429 100644 --- a/src/httpware/decoders/_caching.py +++ b/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,6 +30,7 @@ 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) @@ -35,7 +38,7 @@ def _lookup(cache: "_Cache[V]", model: object, build: Callable[[typing.Any], V | 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 diff --git a/tests/test_decoders_caching.py b/tests/test_decoders_caching.py new file mode 100644 index 0000000..34d196e --- /dev/null +++ b/tests/test_decoders_caching.py @@ -0,0 +1,75 @@ +"""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_releases_old_models() -> None: + decoder = MsgspecDecoder() + models = (msgspec.defstruct(f"S{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_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))) From 548a9ed172680035f9fa52b0d048c34c881eae2e Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sat, 17 Oct 2026 14:51:29 +0300 Subject: [PATCH 2/3] docs: middleware response streams must match the client --- docs/decoders.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/decoders.md b/docs/decoders.md index 831633e..5b0b1da 100644 --- a/docs/decoders.md +++ b/docs/decoders.md @@ -41,7 +41,7 @@ The built-in decoders build a parser once per model (a `pydantic.TypeAdapter` or ## 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 From afeb1a3d171817008013b6b1756b9d257e051795 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sat, 17 Oct 2026 14:55:26 +0300 Subject: [PATCH 3/3] 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. --- tests/test_decoders_caching.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/tests/test_decoders_caching.py b/tests/test_decoders_caching.py index 34d196e..39b74a4 100644 --- a/tests/test_decoders_caching.py +++ b/tests/test_decoders_caching.py @@ -39,12 +39,14 @@ def test_pydantic_cache_is_bounded_and_releases_old_models() -> None: assert refs[0]() is None -def test_msgspec_cache_is_bounded_and_releases_old_models() -> 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)) - refs = _decode_through_fresh_models(decoder, models) + 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 refs[0]() is None + assert all(cached is not models[0] for cached, _ in decoder._adapters.values()) # noqa: SLF001 def test_repr_keyed_cache_is_bounded() -> None: