ClientPayloadError: 400, message: Can not decode content-encoding: zstd - Already at the end of a Zstandard frame

[josumoreno-BP]

Describe the bug

When a server responds with zstd encoded response in multiple frames, current implementation fails decompressing the second frame here. Internally the exception is Already at the end of a Zstandard frame but it ends up raising a ClientPayloadError.

Python docs mention this issue here and suggest to use compression.zstd.decompress to avoid this limitation. I've tested internally with this change and works fine, but I'm not sure if this should be the way to go or you prefer another solution.

I have seen a similar issue into an unrelated project like kafka-python just in case helps understanding it better.

PS: Unfortunately, I won’t be able to open an MR in the next two weeks, so feel free to fix it yourself if you’d like.

To Reproduce

import asyncio

import aiohttp
from compression.zstd import ZstdCompressor

BODY = b"A" * 50_000 + b"B" * 50_000


def zstd_frame(data: bytes) -> bytes:
    c = ZstdCompressor()
    return c.compress(data) + c.flush()


async def main() -> None:
    frame1 = zstd_frame(BODY[:50_000])
    frame2 = zstd_frame(BODY[50_000:])

    async def handle(reader, writer):
        while (await reader.readline()) != b"\r\n":
            pass
        writer.write(
            b"HTTP/1.1 200 OK\r\n"
            b"Content-Type: text/plain\r\n"
            b"Content-Encoding: zstd\r\n"
            b"Transfer-Encoding: chunked\r\n\r\n"
        )
        for frame in (frame1, frame2):
            writer.write(f"{len(frame):x}\r\n".encode() + frame + b"\r\n")
            await writer.drain()
            await asyncio.sleep(0.05)
        writer.write(b"0\r\n\r\n")
        await writer.drain()
        writer.close()

    server = await asyncio.start_server(handle, "127.0.0.1", 0)
    port = server.sockets[0].getsockname()[1]

    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(f"http://127.0.0.1:{port}/") as resp:
                body = await resp.read()
                assert body == BODY
                print("OK")
    except aiohttp.ClientPayloadError as e:
        print(f"FAILED: {type(e).__name__}: {e}")
    finally:
        server.close()
        await server.wait_closed()


if __name__ == "__main__":
    asyncio.run(main())

Expected behavior

No failures on multiframed zstd responses

Logs/tracebacks

FAILED: ClientPayloadError: 400, message:
  Can not decode content-encoding: zstd

Python Version

3.14

aiohttp Version

3.13.3

multidict Version

6.7.1

propcache Version

0.4.1

yarl Version

1.23.0

OS

macOS

Related component

Client

Additional context

No response

Code of Conduct

• I agree to follow the aio-libs Code of Conduct

[Dreamsorcerer]

Python docs mention this issue here and suggest to use compression.zstd.decompress to avoid this limitation

.decompress() has no max_length parameter, so that's not an option.
The docs continue on with "or multiple ZstdDecompressor instances", so that would be the only option we could consider.

[Dreamsorcerer]

I'm struggling to remember any of the details currently, but looking at your repro, it's dealing with chunked messages. Is it expected that every chunk would be compressed separately (I'm guessing so)? If so, then it looks like we need to create a fresh decompressor for each chunk. I'd assume the same applies for brotli and zlib too... (which is fine, as it's all abstracted together)

[Dreamsorcerer]

@josumoreno-BP There have been 2 PRs opened this week. Would be good if you could review and confirm if they fix the problem.

[josumoreno-BP]

Hey @Dreamsorcerer will try between today and tomorrow

[josumoreno-BP]

Hey @Dreamsorcerer I've done a new PR because I was not feeling comfortable with the existing ones. Let me know if you have any questions or disagree on something!