fs: add flush option to createWriteStream()

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

Refs: nodejs#49886
PR-URL: nodejs#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

@@ -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
@@ -2540,6 +2546,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.
@@ -2593,6 +2602,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

lib/fs.js

@@ -3108,6 +3108,7 @@ function createReadStream(path, options) {
  *   emitClose?: boolean;
  *   start: number;
  *   fs?: object | null;
+ *   flush?: boolean;
  *   }} [options]
  * @returns {WriteStream}
  */

lib/internal/fs/promises.js

@@ -350,6 +350,7 @@ class FileHandle extends EventEmitter {
    *   autoClose?: boolean;
    *   emitClose?: boolean;
    *   start: number;
+   *   flush?: boolean;
    *   }} [options]
    * @returns {WriteStream}
    */

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;

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();
+      }));
+    }));
+  });
+});