fix: always terminate active Node Streams (#43072)

`.destroy()` is an important method in the lifecycle of a Node.js
Readable stream. It is typically called to reclaim the resources
(e.g., close file descriptor). The only situations where calling
it manually isn't necessary are when the following events are
emitted first:

- `end`: natural end of a stream
- `error`: stream terminated due to a failure

Prior to this commit the ended state was incorrectly tracked together
with a pending internal error. It led to situations where the request
could get aborted during a read and then get marked as ended (having
pending error).

With this change we disentangle pending "error" and "destroyed" cases to
always properly terminate an active Node.js Readable stream.

Co-authored-by: trop[bot] <37223003+trop[bot]@users.noreply.github.com>
Co-authored-by: Fedor Indutny <79877362+indutny-signal@users.noreply.github.com>

shell/browser/net/node_stream_loader.cc

@@ -43,7 +43,7 @@ NodeStreamLoader::~NodeStreamLoader() {
   }
 
   // Destroy the stream if not already ended
-  if (!ended_) {
+  if (!destroyed_) {
     node::MakeCallback(isolate_, emitter_.Get(isolate_), "destroy", 0, nullptr,
                        {0, 0});
   }
@@ -63,13 +63,21 @@ void NodeStreamLoader::Start(network::mojom::URLResponseHeadPtr head) {
                              std::nullopt);
 
   auto weak = weak_factory_.GetWeakPtr();
-  On("end",
-     base::BindRepeating(&NodeStreamLoader::NotifyComplete, weak, net::OK));
-  On("error", base::BindRepeating(&NodeStreamLoader::NotifyComplete, weak,
-                                  net::ERR_FAILED));
+  On("end", base::BindRepeating(&NodeStreamLoader::NotifyEnd, weak));
+  On("error", base::BindRepeating(&NodeStreamLoader::NotifyError, weak));
   On("readable", base::BindRepeating(&NodeStreamLoader::NotifyReadable, weak));
 }
 
+void NodeStreamLoader::NotifyEnd() {
+  destroyed_ = true;
+  NotifyComplete(net::OK);
+}
+
+void NodeStreamLoader::NotifyError() {
+  destroyed_ = true;
+  NotifyComplete(net::ERR_FAILED);
+}
+
 void NodeStreamLoader::NotifyReadable() {
   if (!readable_)
     ReadMore();
@@ -81,7 +89,7 @@ void NodeStreamLoader::NotifyReadable() {
 void NodeStreamLoader::NotifyComplete(int result) {
   // Wait until write finishes or fails.
   if (is_reading_ || is_writing_) {
-    ended_ = true;
+    pending_result_ = true;
     result_ = result;
     return;
   }
@@ -121,7 +129,7 @@ void NodeStreamLoader::ReadMore() {
     }
 
     readable_ = false;
-    if (ended_) {
+    if (pending_result_) {
       NotifyComplete(result_);
     }
     return;
@@ -146,7 +154,7 @@ void NodeStreamLoader::ReadMore() {
 void NodeStreamLoader::DidWrite(MojoResult result) {
   is_writing_ = false;
   // We were told to end streaming.
-  if (ended_) {
+  if (pending_result_) {
     NotifyComplete(result_);
     return;
   }