fix(decoders): close the response on a jsonl decode or transport error

src/anthropic/_decoders/jsonl.py

@@ -42,22 +42,27 @@ def close(self) -> None:
 
     def __decode__(self) -> Iterator[_T]:
         buf = b""
-        for chunk in self._raw_iterator:
-            for line in chunk.splitlines(keepends=True):
-                buf += line
-                if buf.endswith((b"\r", b"\n", b"\r\n")):
-                    yield construct_type_unchecked(
-                        value=json.loads(buf),
-                        type_=self._line_type,
-                    )
-                    buf = b""
-
-        # flush
-        if buf:
-            yield construct_type_unchecked(
-                value=json.loads(buf),
-                type_=self._line_type,
-            )
+        try:
+            for chunk in self._raw_iterator:
+                for line in chunk.splitlines(keepends=True):
+                    buf += line
+                    if buf.endswith((b"\r", b"\n", b"\r\n")):
+                        yield construct_type_unchecked(
+                            value=json.loads(buf),
+                            type_=self._line_type,
+                        )
+                        buf = b""
+
+            # flush
+            if buf:
+                yield construct_type_unchecked(
+                    value=json.loads(buf),
+                    type_=self._line_type,
+                )
+        finally:
+            # Ensure the response is closed even if the consumer doesn't read all
+            # data, or a JSON decoding or transport-read error is raised above.
+            self.close()
 
     def __next__(self) -> _T:
         return self._iterator.__next__()
@@ -98,22 +103,27 @@ async def close(self) -> None:
 
     async def __decode__(self) -> AsyncIterator[_T]:
         buf = b""
-        async for chunk in self._raw_iterator:
-            for line in chunk.splitlines(keepends=True):
-                buf += line
-                if buf.endswith((b"\r", b"\n", b"\r\n")):
-                    yield construct_type_unchecked(
-                        value=json.loads(buf),
-                        type_=self._line_type,
-                    )
-                    buf = b""
-
-        # flush
-        if buf:
-            yield construct_type_unchecked(
-                value=json.loads(buf),
-                type_=self._line_type,
-            )
+        try:
+            async for chunk in self._raw_iterator:
+                for line in chunk.splitlines(keepends=True):
+                    buf += line
+                    if buf.endswith((b"\r", b"\n", b"\r\n")):
+                        yield construct_type_unchecked(
+                            value=json.loads(buf),
+                            type_=self._line_type,
+                        )
+                        buf = b""
+
+            # flush
+            if buf:
+                yield construct_type_unchecked(
+                    value=json.loads(buf),
+                    type_=self._line_type,
+                )
+        finally:
+            # Ensure the response is closed even if the consumer doesn't read all
+            # data, or a JSON decoding or transport-read error is raised above.
+            await self.close()
 
     async def __anext__(self) -> _T:
         return await self._iterator.__anext__()

tests/decoders/test_jsonl.py

@@ -60,6 +60,27 @@ def body() -> Iterator[bytes]:
     assert await iter_next(iterator) == {"content": "известни"}
 
 
+def test_sync_closes_response_on_decode_error() -> None:
+    response = httpx2.Response(200, stream=httpx2.ByteStream(b"invalid json\n"))
+    decoder = JSONLDecoder(line_type=object, raw_iterator=response.iter_bytes(), http_response=response)
+
+    with pytest.raises(Exception):
+        list(decoder)
+
+    assert response.is_closed
+
+
+async def test_async_closes_response_on_decode_error() -> None:
+    response = httpx2.Response(200, stream=httpx2.ByteStream(b"invalid json\n"))
+    decoder = AsyncJSONLDecoder(line_type=object, raw_iterator=response.aiter_bytes(), http_response=response)
+
+    with pytest.raises(Exception):
+        async for _ in decoder:
+            pass
+
+    assert response.is_closed
+
+
 async def to_aiter(iter: Iterator[bytes]) -> AsyncIterator[bytes]:
     for chunk in iter:
         yield chunk