From d1aa62f1f5462443a5123466d49811b592ed275a Mon Sep 17 00:00:00 2001
From: Colin Ihrig <cjihrig@gmail.com>
Date: Wed, 11 Oct 2023 12:25:05 -0400
Subject: [PATCH] fs: add flush option to createWriteStream()

This commit adds a 'flush' option to the createWriteStream()
family of functions.

Refs: https://github.com/nodejs/node/issues/49886
PR-URL: https://github.com/nodejs/node/pull/50093
Reviewed-By: Yagiz Nizipli <yagiz@nizipli.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Marco Ippolito <marcoippolito54@gmail.com>
---
 doc/api/fs.md                               | 11 +++
 lib/fs.js                                   |  1 +
 lib/internal/fs/promises.js                 |  1 +
 lib/internal/fs/streams.js                  | 28 ++++++-
 test/parallel/test-fs-write-stream-flush.js | 81 +++++++++++++++++++++
 5 files changed, 118 insertions(+), 4 deletions(-)
 create mode 100644 test/parallel/test-fs-write-stream-flush.js

diff --git a/doc/api/fs.md b/doc/api/fs.md
index b15c7be1d1f192..8442fa2a5a18b4 100644
--- a/doc/api/fs.md
+++ b/doc/api/fs.md
@@ -318,6 +318,10 @@ fd.createReadStream({ start: 90, end: 99 });
 
 <!-- YAML
 added: v16.11.0
+changes:
+  - version: REPLACEME
+    pr-url: https://github.com/nodejs/node/pull/50093
+    description: The `flush` option is now supported.
 -->
 
 * `options` {Object}
@@ -326,6 +330,8 @@ added: v16.11.0
   * `emitClose` {boolean} **Default:** `true`
   * `start` {integer}
   * `highWaterMark` {number} **Default:** `16384`
+  * `flush` {boolean} If `true`, the underlying file descriptor is flushed
+    prior to closing it. **Default:** `false`.
 * Returns: {fs.WriteStream}
 
 `options` may also include a `start` option to allow writing data at some
@@ -2514,6 +2520,9 @@ If `options` is a string, then it specifies the encoding.
 <!-- YAML
 added: v0.1.31
 changes:
+  - version: REPLACEME
+    pr-url: https://github.com/nodejs/node/pull/50093
+    description: The `flush` option is now supported.
   - version: v16.10.0
     pr-url: https://github.com/nodejs/node/pull/40013
     description: The `fs` option does not need `open` method if an `fd` was provided.
@@ -2567,6 +2576,8 @@ changes:
   * `fs` {Object|null} **Default:** `null`
   * `signal` {AbortSignal|null} **Default:** `null`
   * `highWaterMark` {number} **Default:** `16384`
+  * `flush` {boolean} If `true`, the underlying file descriptor is flushed
+    prior to closing it. **Default:** `false`.
 * Returns: {fs.WriteStream}
 
 `options` may also include a `start` option to allow writing data at some
diff --git a/lib/fs.js b/lib/fs.js
index 540942d25570b2..5495da5f8f3c00 100644
--- a/lib/fs.js
+++ b/lib/fs.js
@@ -3106,6 +3106,7 @@ function createReadStream(path, options) {
  *   emitClose?: boolean;
  *   start: number;
  *   fs?: object | null;
+ *   flush?: boolean;
  *   }} [options]
  * @returns {WriteStream}
  */
diff --git a/lib/internal/fs/promises.js b/lib/internal/fs/promises.js
index 38ad99db3f040c..11d230077604bf 100644
--- a/lib/internal/fs/promises.js
+++ b/lib/internal/fs/promises.js
@@ -349,6 +349,7 @@ class FileHandle extends EventEmitterMixin(JSTransferable) {
    *   autoClose?: boolean;
    *   emitClose?: boolean;
    *   start: number;
+   *   flush?: boolean;
    *   }} [options]
    * @returns {WriteStream}
    */
diff --git a/lib/internal/fs/streams.js b/lib/internal/fs/streams.js
index 6a213693342b94..43f06d0104de61 100644
--- a/lib/internal/fs/streams.js
+++ b/lib/internal/fs/streams.js
@@ -23,6 +23,7 @@ const {
   kEmptyObject,
 } = require('internal/util');
 const {
+  validateBoolean,
   validateFunction,
   validateInteger,
 } = require('internal/validators');
@@ -92,6 +93,9 @@ const FileHandleOperations = (handle) => {
       PromisePrototypeThen(handle.close(),
                            () => cb(), cb);
     },
+    fsync: (fd, cb) => {
+      PromisePrototypeThen(handle.sync(), () => cb(), cb);
+    },
     read: (fd, buf, offset, length, pos, cb) => {
       PromisePrototypeThen(handle.read(buf, offset, length, pos),
                            (r) => cb(null, r.bytesRead, r.buffer),
@@ -113,14 +117,22 @@ const FileHandleOperations = (handle) => {
 function close(stream, err, cb) {
   if (!stream.fd) {
     cb(err);
-  } else {
-    stream[kFs].close(stream.fd, (er) => {
-      cb(er || err);
+  } else if (stream.flush) {
+    stream[kFs].fsync(stream.fd, (flushErr) => {
+      _close(stream, err || flushErr, cb);
     });
-    stream.fd = null;
+  } else {
+    _close(stream, err, cb);
   }
 }
 
+function _close(stream, err, cb) {
+  stream[kFs].close(stream.fd, (er) => {
+    cb(er || err);
+  });
+  stream.fd = null;
+}
+
 function importFd(stream, options) {
   if (typeof options.fd === 'number') {
     // When fd is a raw descriptor, we must keep our fingers crossed
@@ -350,6 +362,14 @@ function WriteStream(path, options) {
     validateFunction(this[kFs].close, 'options.fs.close');
   }
 
+  this.flush = options.flush;
+  if (this.flush == null) {
+    this.flush = false;
+  } else {
+    validateBoolean(this.flush, 'options.flush');
+    validateFunction(this[kFs].fsync, 'options.fs.fsync');
+  }
+
   // It's enough to override either, in which case only one will be used.
   if (!this[kFs].write) {
     this._write = null;
diff --git a/test/parallel/test-fs-write-stream-flush.js b/test/parallel/test-fs-write-stream-flush.js
new file mode 100644
index 00000000000000..452dd0a1051ab6
--- /dev/null
+++ b/test/parallel/test-fs-write-stream-flush.js
@@ -0,0 +1,81 @@
+'use strict';
+const common = require('../common');
+const tmpdir = require('../common/tmpdir');
+const assert = require('node:assert');
+const fs = require('node:fs');
+const fsp = require('node:fs/promises');
+const test = require('node:test');
+const data = 'foo';
+let cnt = 0;
+
+function nextFile() {
+  return tmpdir.resolve(`${cnt++}.out`);
+}
+
+tmpdir.refresh();
+
+test('validation', () => {
+  for (const flush of ['true', '', 0, 1, [], {}, Symbol()]) {
+    assert.throws(() => {
+      fs.createWriteStream(nextFile(), { flush });
+    }, { code: 'ERR_INVALID_ARG_TYPE' });
+  }
+});
+
+test('performs flush', (t, done) => {
+  const spy = t.mock.method(fs, 'fsync');
+  const file = nextFile();
+  const stream = fs.createWriteStream(file, { flush: true });
+
+  stream.write(data, common.mustSucceed(() => {
+    stream.close(common.mustSucceed(() => {
+      const calls = spy.mock.calls;
+      assert.strictEqual(calls.length, 1);
+      assert.strictEqual(calls[0].result, undefined);
+      assert.strictEqual(calls[0].error, undefined);
+      assert.strictEqual(calls[0].arguments.length, 2);
+      assert.strictEqual(typeof calls[0].arguments[0], 'number');
+      assert.strictEqual(typeof calls[0].arguments[1], 'function');
+      assert.strictEqual(fs.readFileSync(file, 'utf8'), data);
+      done();
+    }));
+  }));
+});
+
+test('does not perform flush', (t, done) => {
+  const values = [undefined, null, false];
+  const spy = t.mock.method(fs, 'fsync');
+  let cnt = 0;
+
+  for (const flush of values) {
+    const file = nextFile();
+    const stream = fs.createWriteStream(file, { flush });
+
+    stream.write(data, common.mustSucceed(() => {
+      stream.close(common.mustSucceed(() => {
+        assert.strictEqual(fs.readFileSync(file, 'utf8'), data);
+        cnt++;
+
+        if (cnt === values.length) {
+          assert.strictEqual(spy.mock.calls.length, 0);
+          done();
+        }
+      }));
+    }));
+  }
+});
+
+test('works with file handles', async () => {
+  const file = nextFile();
+  const handle = await fsp.open(file, 'w');
+  const stream = handle.createWriteStream({ flush: true });
+
+  return new Promise((resolve) => {
+    stream.write(data, common.mustSucceed(() => {
+      stream.close(common.mustSucceed(() => {
+        assert.strictEqual(fs.readFileSync(file, 'utf8'), data);
+        resolve();
+      }));
+    }));
+  });
+});