[core-http] Clean up event listener when streaming is done

[jeremymeng]

Currently our cleanup code removes the abort event listener when the response is
returned. In streaming case even the response is returned, the work is not done
yet. However, with the abort listener removed, we lost the ability to cancel the
streaming.

This change fixes the issue by unregistering the abort listener for
streaming when the stream ends or error happens.

This fixes #12029

[xirzec]

Thanks Jeremy for taking on this work! I have a couple small bits of feedback for your consideration, but I'm eager to find a solution for this problem so that we can also solve it in core-https.

[xirzec]

This seems a little implausible, but given that anything in NodeJS can come in as a stream (even a very simple request body JSON payload), I don't think it's impossible. Is there a reason we can't handle both?

[jeremymeng]

By this rare case I meant, a HTTP request is uploading a stream body while at the same time, streaming down the response body. I chatted with @bterlson and he doesn't think we need to worry about this case.

[jeremymeng]

Oh I think I shouldn't have if-else here. Both uploading stream and downloading stream should have the clean up call back hooked up.

[jeremymeng]

But I need to make sure the last to finish does the clean up.

[jeremymeng]

I remember my intuition was that if for a particular request, if there's a download stream, it would probably ends after the upload stream (if there's any), so clean up at the end of downloading, otherwise clean up at the end of uploading stream if any. Both failing, cleaning up when response returns.

[bterlson]

It's not technically impossible for a response stream to start before a request stream finishes from a conceptual standpoint, but everywhere up and down the stack needs to work in that case and, ultimately, boil up to an application that is comfortable writing a response before a request has even finished. This all seems very rare, and googling seems to confirm.

That said, if it's not hard to support, might not be a bad idea to support it? I imagine a bug here woul dbe very hard to track down...

[jeremymeng]

Because we use the same abort signal for upload stream and download stream, we want to keep the listener around until both stream are done. There are some possible fixes I think:

1. use separate abort signals for upload and download, clean up each one independently, or
2. if we could somehow figure out which stream ends last, do the clean up on that one's end/error event.

[jeremymeng]

Since we don't have to worry about multi-thread race condition, would simple checks like in c2a8a8a work? I am not sure how to test this edge case with real request/response though.

[jeremymeng]

I think the current version should work. @xirzec please have another look.

[xirzec]

So this cleans up the listener - but will cancelation work as expected? Does node-fetch actually stop upload / download streams when the signal is given?

I believe that it could, I just want to make sure we've actually tested it.

[jeremymeng]

Yes I tested it using the storage repo code. node-fetch will destroy request body and emit error to the response body when handling abort event https://github.com/node-fetch/node-fetch/blob/v2.6.1/src/index.js#L60

[xirzec]

Thinking about this more, could we have a solution that did something slightly different, by having the helper return a promise when the stream is completed?

e.g.

let uploadStreamDone = Promise.resolve();
if (httpRequest.abortSignal && isReadableStream(body)) {
   uploadStreamDone = streamIsComplete(body);
}
let downloadStreamDone = Promise.resolve();
if (httpRequest.abortSignal && isReadableStream(operationResponse.readableStreamBody)) {
   downloadStreamDone = streamIsComplete(operationResponse.readableStreamBody);
}

Promise.all([uploadStreamDone, downloadStreamDone]).then(() => {
   httpRequest.abortSignal?.removeEventListener("abort", abortListener);
});

[xirzec]

I don't like how this callback manipulates variables in the outer scope, especially because they are declared below this function definition.

[xirzec]

instead of using a boolean isDownload, what about something like this:

Suggested change

	registerCleanUpListenerOnStream = (stream: NodeJS.ReadableStream, isDownload: boolean) => {
	registerCleanUpListenerOnStream = (stream: NodeJS.ReadableStream, streamType: "download" | "upload") => {

It seems like it would be more readable, since you wouldn't be passing true/false to the method when calling it

[xirzec]

instead of conditionally defining the method and checking its existence later, can't we just check if (httpRequest.abortSignal) instead?

[jeremymeng]

Yes it looks much cleaner with promises. I will see whether it works as expected.

[jeremymeng]

Tested with the repro code and things worked as expected. @xirzec please have another look.

[xirzec]

This looks a lot cleaner! Thanks for updating.

I left one more suggestion that I believe is able to simplify the code a little, but double check that I'm not missing something. 😄

[xirzec]

I think this is identical, but simpler?

Suggested change

	Promise.all([uploadStreamDone, downloadStreamDone])
	.then(() => {
	httpRequest.abortSignal?.removeEventListener("abort", abortListener!);
	return;
	})
	.catch((e) => {
	throw e;
	});
	await Promise.all([uploadStreamDone, downloadStreamDone]);
	httpRequest.abortSignal?.removeEventListener("abort", abortListener!);

[jeremymeng]

This is exactly what I had first, but for mysterious reason it didn't work and repro program exits silently. Let me try again.

[jeremymeng]

Right I had await in fc93036 and my test program exits silently. Using callback works. @bterlson any ideas on the subtle differences between the two that might explain this?

[xirzec]

Hmm well with the callback method, you're no longer participating in the async method that wraps this call, which means you have an unhandled promise rejection (which in some NodeJS will crash your entire app.)

You don't actually want to do that, I think, since this method is already returning a promise. However, I'm not sure how await works inside finally... does it override the return/reject state of the promise?

[xirzec]

Looks like await inside finally should do the right thing (throw if there is an error)

[jeremymeng]

I did some more experiments. This silent exiting behavior seems to be related to how the helper is implemented. when await Promise.all() is used, if I don't have the empty "data" event handler the test program exits silently.

 function isStreamComplete(stream: Readable): Promise<void> {
   return new Promise((resolve) => {
     stream.on("data", () => {
       return;
     })
     stream.on("close", () => { console.log("### isStreamComplete on close"); resolve(); });
     stream.on("error", () => { console.log("### isStreamComplete on error"); });
     stream.on("end",   () => { console.log("### isStreamComplete on end");   });
   });
 }

Also I found that with await Promise.all() in our finally block, we are blocking test program's execution until the streaming is finished. User code of stream.destroy() on the download stream after blobClient.download() call for example won't get executed until download stream is complete, which is wrong.

testing 0
### isStreamComplete on end
### isStreamComplete on close
download call done for 0
destroying stream 0...

Comparing to the result of using Promise.all().then() callback: the stream can be destroyed, and then the promise resolves.

testing 0
download call done for 0
destroying stream 0...
### isStreamComplete on end
stream 0 on end
### isStreamComplete on close
stream 0 closed

I still don't understand why an empty "data" event handler matters. Regardless, we shouldn't block on stream completion.

[xirzec]

Ah yes, you are absolutely right that we shouldn't block on completion. I think the only dilemma we have here is what to do when the new promise rejects. We can't throw/leave it rejected since the consumer has no way to handle it.

Should we log or ignore it?

[jeremymeng]

Would it be fine if our promise helper always resolves and never rejects? The doc says

A Readable stream will always emit the 'close' event if it is created with the emitClose option.

Regarding the test program silent exiting when await Promse.all() is used. @bterlson helped investigate it and found out that if we await on promise that never resolves or rejects, NodeJS exits the process with code 13

• 13 Unfinished Top-Level Await: await was used outside of a function in the top-level code, but the passed Promise never resolved.

[xirzec]

I'm fine with doing Promise.all().then(cleanup).catch(() => { /* ignore errors */ })

[xirzec]

I'm flipping my vote back until we figure out about the unhandled rejection issue, since that's caused distress to customers before

[jeremymeng]

/azp run js - storage-blob - tests

[azure-pipelines]

Azure Pipelines successfully started running 1 pipeline(s).

[jeremymeng]

I'm flipping my vote back until we figure out about the unhandled rejection issue, since that's caused distress to customers before

@xirzec please have another look. I ported the fix to core-https.

[xirzec]

I'm really happy where this change landed. We should make sure to ship core-http again in February so this goes out.

[xirzec]

is this unrelated?

[jeremymeng]

It's related. node-fetch seems to choose a different abort error message than browser...my commit message from f99cefc. Previously a different AbortError is caught.

Now that aborting streaming is working, we are getting AbortError from
node-fetch which use a different message "The user aborted a request."
than the browser fetch API does. so stop verifying error.message.

[xirzec]

thanks for explaining

[xirzec]

Maybe a teensy more detail for folks who have to ponder the log message?

Suggested change

	logger.warning("Error when cleaning up abortListener", e);
	logger.warning("Error when cleaning up abortListener on httpRequest", e);

[jeremymeng]

Certainly, updated.

[jeremymeng]

@ljian3377 please have a look at the storage test updates.

[ljian3377]

Storage side change looks good to me.