CompressionLayer does not produce compressed data in streaming

[Geal]

Bug Report

Version

0.3.4

Platform

Linux 5.18.10-76051810-generic, but probably not relevant

Crates

tower-http, async-compression

Description

We noticed in https://github.com/apollographql/router/issues/1572 that the CompressionLayer waits until the entire response body is compressed to send it to the client. This is due to the Encoder behaviour in async-compression: https://github.com/Nemo157/async-compression/issues/154.

How we saw that result:

• the router produces a multipart response, where we want the client to receive parts as soon as they are produced. The parts are produced as a Stream of Bytes that is wrapped in an axum StreamBody
• In one test the first part was produced immediately, and the next one after a few seconds
• If the response is not compressed, we see them come as soon as possible
• If the response is compressed (changing the compression algorithm does not change anything), then we receive nothing, and after a few seconds we receive both parts at the same time

The issue comes from this part:

https://github.com/Nemo157/async-compression/blob/ada65c660bcea83dc6a0c3d6149e5fbcd039f739/src/tokio/bufread/generic/encoder.rs#L63-L74

When the underlying stream returns Poll::Pending, ready! will return it directly, so no data will be produced. I patched this in our router to produce data whenever we see a Pending, but that's not a proper solution.

Instead, the CompressionLayer should be able to direct the Encoder to produce data depending on conditions like how much data could be produced, or how long since the last chunk was sent

[davidpdrsn]

Thanks for the report! Compressing streaming response is hard and I don't know a lot about how to do it or whether it requires changes in async-compression.

I don't have time to work on this but PRs is much appreciated!

[Geal]

@davidpdrsn in https://github.com/Nemo157/async-compression/issues/154 we're going towards adding a poll_flush method to the encoders, but I have difficulties wrapping my head around WrapBody. It seems very generic, used for both compression and decompression, and since it uses ReaderStream I do not see a point where calling poll_flush would fit.

Would it be acceptable to have something else than WrapBody for the compression side?

I'm also wondering if this issue would also appear on the decompression side, considering https://github.com/Nemo157/async-compression/issues/123

[davidpdrsn]

Would it be acceptable to have something else than WrapBody for the compression side?

Yeah the whole WrapBody thing is pretty complex. It might be pulling its weight. If you wanna take a stab at refactoring things to make poll_flush possible to access then please do!