lib: fix unhandled errors in webstream adapters

WebStream's Readable controller does not tolerate `.close()` being
called after an `error`. However, when wrapping a Node's Readable stream
it is possible that the sequence of events leads to `finished()`'s
callback being invoked after such `error`.
In order to handle this, in this change we call the `finished()` handler
earlier when controller is canceled, and always handle this as an error
case.
Fix: #54205
PR-URL: #54206Fixes: #54205
Reviewed-By: Yagiz Nizipli <yagiz@nizipli.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Mattias Buelens <mattias@buelens.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>

Author: indutny

lib/internal/webstreams/adapters.js

@@ -424,6 +424,7 @@ function newReadableStreamFromStreamReadable(streamReadable, options = kEmptyObj
 conststrategy=evaluateStrategyOrFallback(options?.strategy);
 
 letcontroller;
+letwasCanceled=false;
 
 functiononData(chunk){
 // Copy the Buffer to detach it from the pool.
@@ -448,6 +449,10 @@ function newReadableStreamFromStreamReadable(streamReadable, options = kEmptyObj
 streamReadable.on('error',()=>{});
 if(error)
 returncontroller.error(error);
+// Was already canceled
+if(wasCanceled){
+return;
+}
 controller.close();
 });
 
@@ -459,6 +464,7 @@ function newReadableStreamFromStreamReadable(streamReadable, options = kEmptyObj
 pull(){streamReadable.resume();},
 
 cancel(reason){
+wasCanceled=true;
 destroy(streamReadable,reason);
 },
 },strategy);

test/parallel/test-stream-readable-from-web-termination.js

@@ -0,0 +1,15 @@
+'use strict';
+require('../common');
+const{ Readable }=require('stream');
+
+{
+constr=Readable.from(['data']);
+
+constwrapper=Readable.fromWeb(Readable.toWeb(r));
+
+wrapper.on('data',()=>{
+// Destroying wrapper while emitting data should not cause uncaught
+// exceptions
+wrapper.destroy();
+});
+}

test/parallel/test-stream-readable-to-web-termination.js

@@ -0,0 +1,12 @@
+'use strict';
+require('../common');
+const{ Readable }=require('stream');
+
+{
+constr=Readable.from([]);
+// Cancelling reader while closing should not cause uncaught exceptions
+r.on('close',()=>reader.cancel());
+
+constreader=Readable.toWeb(r).getReader();
+reader.read();
+}