refactor(body_cap): share non-iteration logic between sync/async

planning/changes/2026-07-13.06-body-cap-shared-non-iteration-logic.md

@@ -0,0 +1,73 @@
+---
+summary: Share the byte-identical declared-length-reject and Response-reconstruction blocks between _read_capped and _read_capped_async.
+---
+
+# Change: Share body_cap's non-iteration logic between sync/async
+
+**Lane:** lightweight — single file, no new file, no public-API change, no
+test changes (existing suite is the parity net).
+
+## Goal
+
+`_read_capped` and `_read_capped_async` (`src/httpware/_internal/body_cap.py`)
+hand-duplicate two byte-identical blocks: the declared-`Content-Length`
+early-reject, and the final `httpx2.Response` reconstruction. Only the
+actual byte-accumulation loop between them genuinely differs (sync uses
+`_accumulate_capped`+`_CapExceeded`; async inlines its own loop) and stays
+duplicated — same judgment already applied to every other sync/async split
+this session. No behavior change.
+
+## Approach
+
+Two new module-level functions, local to `body_cap.py` (used only by its
+own two functions — same precedent as every other local-scope decision
+this session):
+
+```python
+def _reject_if_declared_length_exceeds_cap(response: httpx2.Response, cap: int) -> int | None:
+    """Return the parsed Content-Length, raising ResponseTooLargeError(reason="declared") if it already exceeds cap."""
+    content_length = _parse_content_length(response.headers.get("content-length"))
+    if content_length is not None and content_length > cap:
+        raise ResponseTooLargeError(
+            status_code=response.status_code, limit=cap, content_length=content_length, reason="declared"
+        )
+    return content_length
+
+
+def _build_buffered_response(response: httpx2.Response, content: bytes, request: httpx2.Request) -> httpx2.Response:
+    """Rebuild a buffered Response from the original streaming `response` and the final decoded `content`."""
+    return httpx2.Response(
+        status_code=response.status_code,
+        headers=_buffered_headers(response.headers),
+        content=content,
+        request=request,
+        extensions=_safe_extensions(response.extensions),
+        history=response.history,
+    )
+```
+
+`content_length` is returned (not just checked) because both call sites
+reuse it in their second (`reason="streamed"`) raise. Both `_read_capped`
+and `_read_capped_async` call these two functions in place of their inline
+blocks; the accumulation loop between them is untouched.
+
+## Files
+
+- `src/httpware/_internal/body_cap.py` — add the two functions; replace
+  the duplicated declared-reject block and the duplicated
+  `httpx2.Response(...)` reconstruction in both `_read_capped` and
+  `_read_capped_async` with calls to them.
+- No test file changes — `tests/test_body_cap.py` already exercises both
+  extraction targets across both sync and async (declared-over-cap,
+  streamed-over-cap using the returned `content_length`, within-cap /
+  exact-cap / empty-body using the reconstruction path).
+
+## Verification
+
+- [ ] `just test` — full suite green, 100% coverage maintained, same test
+      count (pure refactor, no new/removed tests).
+- [ ] `just lint` — clean.
+- [ ] Grep `body_cap.py` to confirm the declared-reject block and the
+      `httpx2.Response(...)` construction each appear exactly once (inside
+      the two new functions), not inline in `_read_capped`/
+      `_read_capped_async`.

src/httpware/_internal/body_cap.py

@@ -87,6 +87,28 @@ def _response_has_body(method: str, status_code: int) -> bool:
     return method.upper() != "HEAD" and status_code not in _BODILESS_STATUS
 
 
+def _reject_if_declared_length_exceeds_cap(response: httpx2.Response, cap: int) -> int | None:
+    """Return the parsed Content-Length, raising ResponseTooLargeError(reason="declared") if it already exceeds cap."""
+    content_length = _parse_content_length(response.headers.get("content-length"))
+    if content_length is not None and content_length > cap:
+        raise ResponseTooLargeError(
+            status_code=response.status_code, limit=cap, content_length=content_length, reason="declared"
+        )
+    return content_length
+
+
+def _build_buffered_response(response: httpx2.Response, content: bytes, request: httpx2.Request) -> httpx2.Response:
+    """Rebuild a buffered Response from the original streaming `response` and the final decoded `content`."""
+    return httpx2.Response(
+        status_code=response.status_code,
+        headers=_buffered_headers(response.headers),
+        content=content,
+        request=request,
+        extensions=_safe_extensions(response.extensions),
+        history=response.history,
+    )
+
+
 def _read_capped(response: httpx2.Response, cap: int, request: httpx2.Request) -> httpx2.Response:
     """Buffer a streaming sync `response` under `cap` decoded bytes; return a buffered Response.
 
@@ -98,49 +120,27 @@ def _read_capped(response: httpx2.Response, cap: int, request: httpx2.Request) -
     if not _response_has_body(request.method, response.status_code):
         response.read()  # empty body; preserve the original response (and its headers)
         return response
-    content_length = _parse_content_length(response.headers.get("content-length"))
-    if content_length is not None and content_length > cap:
-        raise ResponseTooLargeError(
-            status_code=response.status_code, limit=cap, content_length=content_length, reason="declared"
-        )
+    content_length = _reject_if_declared_length_exceeds_cap(response, cap)
     try:
         content = _accumulate_capped(response.iter_bytes(), cap)
     except _CapExceeded:
         raise ResponseTooLargeError(
             status_code=response.status_code, limit=cap, content_length=content_length, reason="streamed"
         ) from None
-    return httpx2.Response(
-        status_code=response.status_code,
-        headers=_buffered_headers(response.headers),
-        content=content,
-        request=request,
-        extensions=_safe_extensions(response.extensions),
-        history=response.history,
-    )
+    return _build_buffered_response(response, content, request)
 
 
 async def _read_capped_async(response: httpx2.Response, cap: int, request: httpx2.Request) -> httpx2.Response:
     """Async mirror of `_read_capped` (counts decoded bytes from `aiter_bytes`)."""
     if not _response_has_body(request.method, response.status_code):
         await response.aread()  # empty body; preserve the original response (and its headers)
         return response
-    content_length = _parse_content_length(response.headers.get("content-length"))
-    if content_length is not None and content_length > cap:
-        raise ResponseTooLargeError(
-            status_code=response.status_code, limit=cap, content_length=content_length, reason="declared"
-        )
+    content_length = _reject_if_declared_length_exceeds_cap(response, cap)
     buf = bytearray()
     async for chunk in response.aiter_bytes():
         buf += chunk
         if len(buf) > cap:
             raise ResponseTooLargeError(
                 status_code=response.status_code, limit=cap, content_length=content_length, reason="streamed"
             )
-    return httpx2.Response(
-        status_code=response.status_code,
-        headers=_buffered_headers(response.headers),
-        content=bytes(buf),
-        request=request,
-        extensions=_safe_extensions(response.extensions),
-        history=response.history,
-    )
+    return _build_buffered_response(response, bytes(buf), request)

tests/test_body_cap.py

@@ -125,6 +125,7 @@ def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
         with pytest.raises(ResponseTooLargeError) as caught:
             _read_capped(resp, 1000, resp.request)
         assert caught.value.reason == "streamed"  # compressed CL (small) passed; decoded tripped
+        assert caught.value.content_length == len(raw)  # the declared (compressed) length, threaded through
     finally:
         resp.close()
         client.close()
@@ -192,6 +193,23 @@ def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
         await client.aclose()
 
 
+async def test_read_capped_async_gzip_bomb_trips_on_decoded_bytes() -> None:
+    raw = gzip.compress(b"A" * 100_000)
+
+    def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
+        return httpx2.Response(200, headers={"content-encoding": "gzip"}, content=raw)
+
+    client, resp = await _async_stream(handler)
+    try:
+        with pytest.raises(ResponseTooLargeError) as caught:
+            await _read_capped_async(resp, 1000, resp.request)
+        assert caught.value.reason == "streamed"  # compressed CL (small) passed; decoded tripped
+        assert caught.value.content_length == len(raw)  # the declared (compressed) length, threaded through
+    finally:
+        await resp.aclose()
+        await client.aclose()
+
+
 async def test_read_capped_async_head_with_large_declared_length_not_rejected() -> None:
     def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
         return httpx2.Response(200, headers={"content-length": "50000000"})