HTTP Streams, not possible to implement backpressure or to detect when a client disconnects

[ftokarev]

I am writing a function that retrieves the files from Azure Blob Storage and returns a zip file containing them. HTTP Streams seem like a perfect match for this use case: I can just start loading files one by one from storage and stream the archive to the client on the fly.

The proof of concept works, but I'm facing the following issues:

1. There seems to be no way to detect that a client closed the connection. The function keeps writing data somewhere, and only when everything is written, there appears an error in the log ([Error] Executed '<function name>' (Failed, Id=<id>, Duration=54018ms))
2. There is no way to implement backpressure. The "drain" event is never emitted on the stream that I return in the response body, the data is consumed immediately and fully whenever I pass it into the stream. I tested with slow clients: the function keeps sending data, while the memory used by the function app rapidly increases. Out of curiosity I looked around the codebase, it seems the reason is this code, where the response is written unconditionally, without checking the value returned by ServerResponse.write.

For completeness, this is the PoC code I have:

import { InvocationContext, HttpHandler, HttpRequest } from '@azure/functions';
import * as archiver from "archiver";
import * as stream from "stream";

export const Download: HttpHandler = async (request: HttpRequest, context: InvocationContext) => {
    const ptStream = new stream.PassThrough({ highWaterMark: 64 * 1024 });

    ptStream.on("drain", () => {
        context.log("This is never called");
    });

    const archive = archiver("zip", { zlib: { level: 1 } });
    archive.pipe(ptStream);

    const queue = [...input.filenames];

    const processNext = async () => {
        const filename = queue.pop();
        const blockBlobClient = containerClient.getBlockBlobClient(filename);
        const downloadResponse = await blockBlobClient.download();
        const stream = downloadResponse.readableStreamBody as NodeJS.ReadableStream;

        archive.append(stream, { name: filename });
    };

    archive.on('entry', (e) => {
        if (queue.length == 0) {
            return archive.finalize();
        }

        processNext();
    });

    processNext();

    return {
        body: ptStream,
        status: 200,
        headers: {
            "Content-Type": "application/zip",
            "Content-Disposition": `attachment; filename=${input.archiveName}`
        }
    }
}

[ftokarev]

For the second issue, perhaps the fix can be as simple as replacing this code with the following?

for await (const chunk of userRes.body.values()) {
    const canWrite = proxyRes.write(chunk);

    if (!canWrite) {
        await new Promise((resolve) => proxyRes.once("drain", resolve));
    }
}

[ejizba]

Hi @ftokarev thanks for the detailed report and suggested fix - we'll look into it

[DawidRubch]

Hey, any updates on this? I'm struggiling with the cancel stream part. I might have to move my entire API to somewhere else if that's not possible

[DawidRubch]

Or maybe(if possible) is there a way to cancel a running request by having the invocationId?

That could do it, not ideal approach.