What is the correct way to decompress a file and return?

[nutabi]

Hi, I'm implementing two handlers. One is for user to upload files, and the other is to download them.

The upload handler seems fine; however, its output, when decompressed using zlib-flate, gives zlib-flate: WARNING: zlib code -5, msg = input stream is complete but output may still be valid. The output is in fact okay to me.

However, the download handler cannot decompress the file with error connection error: error from user's HttpBody stream: deflate decompression error.

When the compression/decompression streams are commented out, the results are okay. Still, since I might be storing large files, I'd love to have compression supported.

These are the code snippets for the handlers:

1. Upload

let mut file = tokio::fs::File::create(path).await?;
let mut file = async_compression::tokio::write::ZlibEncoder::new(file); // Comment out to disable compression.
let mut stream = stream.map_err(...); // stream is BodyStream
let mut stream = StreamReader::new(stream);
io::copy(&mut stream, &mut file).await?;

2. Download

let path = ...;
let buf = tokio::fs::File::open(path).await?;
let buf = tokio::io::BufReader::new(file);  // Comment out to disable decompression.
let buf = async_compression::tokio::bufread::ZlibDecoder::new(buf); // Comment out to disable decompression.
let buf = tokio_util::io::ReaderStream::new(buf);
let buf = axum::body::StreamBody::new(stream);
return Ok(buf);

(See here if you need the full code for diagnostic.)

Thank you!

[davidpdrsn]

I'm not sure but you should be able to reproduce this without using axum, just using streams directly.

[davidpdrsn]

The code you show doesn’t use axum except for the StreamBody. I’m saying try to make it work without axum. That’ll be easier to debug.

[nutabi]

My apologies maybe the code snippets weren’t clear, but I do use axum for this. The upload handler takes in a BodyStream, whereas the download handler needs to return an impl IntoResponse.

[davidpdrsn]

Yes I know. What I'm saying is try to reproduce this without axum.

[nutabi]

Sorry, I get what you meant now. I'll try and get back.

[nutabi]

Hi @davidpdrsn, I have not managed to reproduce this error without using axum with the following code:

use async_compression::tokio::{write::ZlibDecoder, bufread::ZlibEncoder};
use tokio::{fs::File, io::BufReader};

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // 1. Compression.
    let file = File::open("Cargo.toml").await?;
    let file = BufReader::new(file);
    let mut encoder = ZlibEncoder::new(file);
    let mut file = File::create("Cargo.toml.zl").await?;
    tokio::io::copy(&mut encoder, &mut file).await?;

    // 2. Decompression.
    let file = File::create("Cargo.toml.out").await?;
    let mut decoder = ZlibDecoder::new(file);
    let mut file = File::open("Cargo.toml.zl").await?;
    tokio::io::copy(&mut file, &mut decoder).await?;

    Ok(())
}

I have gone back to my handling code and taken a look. I did made a few mistakes, but even after fixing it, it is still not working. Here's the new code:

// Create output file writer.
//
// First, we generate path to file from paste's UUID and create the file.
//
// Then, we wrap a BufWriter then a ZlibEncoder to compress the content.
let path = generate_path(paste_record.paste_id).await?;
let out_file = fs::File::create(path.clone())
    .await
    .context("Unable to create Pastebin file")?;
let mut encoder = Compressor::new(out_file);

// Create input stream reader.
//
// We need to convert BodyStream in to a StreamReader. However, we need to
// map BodyStream's error to tokio::io::Error first, then we wrap a
// StreamReader afterwards.
let stream = stream.map_err(|e| io::Error::new(io::ErrorKind::Other, e));
let mut stream = StreamReader::new(stream);

// Copy from stream to encoder.
if let Err(e) = io::copy(&mut stream, &mut encoder).await { ... }

And:

// Create input file reader.
//
// Similar to compresison, we first generate path from UUID then open the
// file, instead of creating it since we want to read from it.
//
// Then, we wrap a BufReader then a ZlibDecoder to decompress the content.
let path = generate_path(paste_record.paste_id).await?;
let in_file = fs::File::open(path)
    .await
    .context("Unable to open paste file")?;
let in_file = BufReader::new(in_file);
let decoder = Decompressor::new(in_file);

// Create output stream writer
//
// Unlike compression, we don't have to use `tokio::io::copy` as we cannot
// create an underlying writer. Hence, we will just build on top of the
// decoder above.
//
// We first wrap a ReaderStream then a StreamBody to convert it to
// `impl Response`.

// `StreamBody` needs `Stream`, so we then convert `AsyncRead` to `Stream`
// using a `ReaderStream`, then we convert the whole thing to `StreamBody`.
let stream = ReaderStream::new(decoder);
let body = StreamBody::new(stream);

Please advise :)

Edit: The problem with the reproduction code is that the decompression flow is not exactly the same. I cannot create an empty BodyStream then use tokio::io::copy to copy from reader to writer. I guess the problem is around there.

Another edit: I changed the reproduction code to inspect the compressed data. Unlike the upload handler, zlib-flate does not produce a warning when I decompress the output file Cargo.toml.zl, so I believe the problem is with the upload handler not flushed before it returns, and async-compression takes the warning as an error. No idea how to verify this hypothesis though.