From ad5377c09ecbe3802577dbd7b1a48209bd912731 Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Wed, 29 Jul 2026 21:54:34 +0000
Subject: [PATCH] fs.Utf8Stream: don't emit post-reopen 'drain' while a write
 is in flight

In async mode, fileOpened() emits 'ready' synchronously after a reopen.
If a 'ready' listener calls write(), #actualWrite starts an async
fs.write and sets #writing=true. The subsequent '!#writing && #len >
#minLength' check is false, so the code fell through to the 'reopening'
branch and scheduled process.nextTick(emit('drain')). That drain fires
before the fs.write completes, so a once('drain') listener that reads
the reopened file back sees an empty file.

This is the source of test-fastutf8stream-reopen.js flaking on ~30% of
CI builds: under threadpool contention Bun's fs.write physical write
completes after nextTick, exposing the premature drain. (Node v26.3.0
has the identical upstream bug; it just doesn't surface under Node's
libuv timing.)

Guard the post-reopen drain on !#writing so #release handles the drain
when the pending write completes.
---
 src/js/internal/streams/fast-utf8-stream.ts   |  5 +-
 .../node/fs/utf8stream-reopen-drain.test.ts   | 68 +++++++++++++++++++
 2 files changed, 72 insertions(+), 1 deletion(-)
 create mode 100644 test/js/node/fs/utf8stream-reopen-drain.test.ts

diff --git a/src/js/internal/streams/fast-utf8-stream.ts b/src/js/internal/streams/fast-utf8-stream.ts
index c647614428c4..2ea0e5a16072 100644
--- a/src/js/internal/streams/fast-utf8-stream.ts
+++ b/src/js/internal/streams/fast-utf8-stream.ts
@@ -454,7 +454,10 @@ class Utf8Stream extends EventEmitter {
       // start
       if ((!this.#writing && this.#len > this.#minLength) || this.#flushPending) {
         this.#actualWrite();
-      } else if (reopening) {
+      } else if (reopening && !this.#writing) {
+        // A 'ready' listener may have called write() synchronously and started
+        // an async fs.write; emitting 'drain' now would signal completion while
+        // that write is still in flight. #release emits 'drain' when it lands.
         process.nextTick(() => this.emit("drain"));
       }
     };
diff --git a/test/js/node/fs/utf8stream-reopen-drain.test.ts b/test/js/node/fs/utf8stream-reopen-drain.test.ts
new file mode 100644
index 000000000000..9e74c89ef401
--- /dev/null
+++ b/test/js/node/fs/utf8stream-reopen-drain.test.ts
@@ -0,0 +1,68 @@
+import { expect, test } from "bun:test";
+import { bunEnv, bunExe, tempDir } from "harness";
+
+// Deflake for test/js/node/test/parallel/test-fastutf8stream-reopen.js.
+//
+// After reopen(), fileOpened() emits 'ready' synchronously (async mode). If a
+// 'ready' listener calls write(), #actualWrite starts an async fs.write and
+// sets #writing=true. fileOpened() must then NOT schedule its post-reopen
+// nextTick 'drain': that drain would fire while the write is still in flight,
+// so a listener that reads the file on 'drain' observes an empty file. The
+// real flake surfaces when Bun's fs.write threadpool task completes after
+// nextTick under CI load; here we force it deterministically by deferring the
+// write callback with setImmediate.
+test("fs.Utf8Stream: reopen does not emit 'drain' while a write started from 'ready' is pending", async () => {
+  using dir = tempDir("utf8stream-reopen-drain", {});
+  await using proc = Bun.spawn({
+    cmd: [
+      bunExe(),
+      "-e",
+      `
+const { Utf8Stream } = require("node:fs");
+const fs = require("node:fs");
+const path = require("node:path");
+const assert = require("node:assert");
+
+const dest = path.join(process.env.DIR, "a.log");
+let afterReopen = false;
+const slowFs = {
+  write(fd, data, enc, cb) {
+    if (afterReopen) {
+      setImmediate(() => fs.write(fd, data, enc, cb));
+    } else {
+      fs.write(fd, data, enc, cb);
+    }
+  },
+};
+
+const stream = new Utf8Stream({ dest, sync: false, fs: slowFs });
+assert.ok(stream.write("hello world\\n"));
+assert.ok(stream.write("something else\\n"));
+const after = dest + "-moved";
+
+stream.once("drain", () => {
+  fs.renameSync(dest, after);
+  stream.reopen();
+  stream.once("ready", () => {
+    afterReopen = true;
+    assert.ok(stream.write("after reopen\\n"));
+    stream.once("drain", () => {
+      assert.strictEqual(stream.writing, false, "'drain' fired while a write is in flight");
+      assert.strictEqual(fs.readFileSync(after, "utf8"), "hello world\\nsomething else\\n");
+      assert.strictEqual(fs.readFileSync(dest, "utf8"), "after reopen\\n");
+      stream.end();
+      stream.on("close", () => console.log("PASS"));
+    });
+  });
+});
+`,
+    ],
+    env: { ...bunEnv, DIR: String(dir) },
+    stdout: "pipe",
+    stderr: "pipe",
+  });
+  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+  expect(stderr).toBe("");
+  expect(stdout).toBe("PASS\n");
+  expect(exitCode).toBe(0);
+});