stream: sync stream unpipe resume

pipe() ondata should not control flow state if cleaned up.

Fixes: nodejs#31190

lib/_stream_readable.js

@@ -739,6 +739,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
           debug('false write response, pause', state.awaitDrainWriters.size);
           state.awaitDrainWriters.add(dest);
         }
+        src.pause();
       }
       if (!ondrain) {
         // When the dest drains, it reduces the awaitDrain counter
@@ -748,7 +749,6 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
         ondrain = pipeOnDrain(src, dest);
         dest.on('drain', ondrain);
       }
-      src.pause();
     }
   }
 

test/parallel/test-stream-readable-unpipe-resume.js

@@ -0,0 +1,21 @@
+'use strict';
+
+const common = require('../common');
+
+const stream = require('stream')
+
+const fs = require('fs');
+const readStream = fs.createReadStream('out/Release/node')
+
+const transformStream = new class extends stream.Transform {
+  _transform() {
+    readStream.unpipe()
+    readStream.resume()
+  }
+}
+
+readStream.on('end', common.mustCall());
+
+readStream
+  .pipe(transformStream)
+  .resume()