stream: fix early drain after Utf8Stream reopen

If a 'ready' listener starts a write after reopen(), the reopen path
still emits 'drain' from a nextTick before that write completes, so a
listener reading the file on 'drain' can observe it empty. This is the
race behind the test-fastutf8stream-reopen flake deflaked on the test
side in 2e8a4b1.

Skip the extra emit when a write is in flight: #release() emits the
real 'drain' once the write completes, so no event is lost.

The regression test defers the reopened file's fs.write by one
setImmediate, deterministically landing the write after the nextTick
on which the premature 'drain' used to fire.

Refs: 2e8a4b1a8ce
Signed-off-by: Matteo Collina <hello@matteocollina.com>

Author: mcollina

lib/internal/streams/fast-utf8-stream.js

@@ -500,7 +500,9 @@ class Utf8Stream extends EventEmitter {
       // start
       if ((!this.#writing && this.#len > this.#minLength) || this.#flushPending) {
         this.#actualWrite();
-      } else if (reopening) {
+      } else if (reopening && !this.#writing) {
+        // Do not emit 'drain' if a 'ready' listener started a write:
+        // #release() will emit the real 'drain' when that write completes.
         process.nextTick(() => this.emit('drain'));
       }
     };

test/parallel/test-fastutf8stream-reopen.js

@@ -8,6 +8,7 @@ const {
   openSync,
   readFile,
   renameSync,
+  write,
 } = require('node:fs');
 const { Utf8Stream } = require('node:fs');
 const { join } = require('node:path');
@@ -27,6 +28,47 @@ function getTempFile() {
 runTests(false);
 runTests(true);
 
+// A write started by a 'ready' listener after reopen() must complete before
+// 'drain' is emitted. Deferring the reopened file's write by one setImmediate
+// makes the write land after the nextTick on which reopen() used to emit a
+// premature 'drain'. Async mode only: sync mode writes before 'ready'.
+{
+  const dest = getTempFile();
+  const after = dest + '-new';
+  const stream = new Utf8Stream({
+    dest,
+    minLength: 0,
+    sync: false,
+    fs: {
+      write(fd, buf, enc, cb) {
+        if (stream.file === after) {
+          setImmediate(() => write(fd, buf, enc, cb));
+          return;
+        }
+        return write(fd, buf, enc, cb);
+      },
+    },
+  });
+
+  assert.ok(stream.write('hello world\n'));
+
+  stream.once('drain', common.mustCall(() => {
+    stream.reopen(after);
+
+    stream.once('ready', common.mustCall(() => {
+      assert.ok(stream.write('after reopen\n'));
+
+      stream.once('drain', common.mustCall(() => {
+        assert.strictEqual(stream.writing, false);
+        readFile(after, 'utf8', common.mustSucceed((data) => {
+          assert.strictEqual(data, 'after reopen\n');
+          stream.end();
+        }));
+      }));
+    }));
+  }));
+}
+
 function runTests(sync) {
 
   {