vfs: write RealFSProvider files to open fd

lib/internal/vfs/providers/real.js

@@ -2,6 +2,7 @@
 
 const {
   ArrayPrototypePush,
+  MathMin,
   Promise,
   StringPrototypeStartsWith,
 } = primordials;
@@ -11,11 +12,21 @@ const fs = require('fs');
 const path = require('path');
 const { VirtualProvider } = require('internal/vfs/provider');
 const { VirtualFileHandle } = require('internal/vfs/file_handle');
-const { getValidatedPath } = require('internal/fs/utils');
+const {
+  constants: { kWriteFileMaxChunkSize },
+  getOptions,
+  getValidatedPath,
+} = require('internal/fs/utils');
+const { isIterable } = require('internal/streams/utils');
+const { isArrayBufferView } = require('internal/util/types');
 const { setOwnProperty } = require('internal/util');
+const { parseFileMode, validateBoolean } = require('internal/validators');
 const {
-  ERR_METHOD_NOT_IMPLEMENTED,
-} = require('internal/errors').codes;
+  AbortError,
+  codes: {
+    ERR_METHOD_NOT_IMPLEMENTED,
+  },
+} = require('internal/errors');
 const {
   createEACCES,
   createEBADF,
@@ -24,6 +35,16 @@ const {
 
 const kReadFileUnknownBufferLength = 8192;
 
+function isCustomIterable(obj) {
+  return isIterable(obj) && !isArrayBufferView(obj) && typeof obj !== 'string';
+}
+
+function checkAborted(signal) {
+  if (signal?.aborted) {
+    throw new AbortError(undefined, { cause: signal.reason });
+  }
+}
+
 /**
  * A file handle that wraps a real file descriptor.
  */
@@ -33,7 +54,6 @@ const kReadFileUnknownBufferLength = 8192;
 // sync-opened handles can still share one underlying handle for async ops.
 class RealFileHandle extends VirtualFileHandle {
   #fd;
-  #realPath;
 
   #checkClosed(syscall) {
     if (this.closed) {
@@ -60,12 +80,10 @@ class RealFileHandle extends VirtualFileHandle {
    * @param {string} flags The open flags
    * @param {number} mode The file mode
    * @param {number} fd The real file descriptor
-   * @param {string} realPath The real filesystem path
    */
-  constructor(path, flags, mode, fd, realPath) {
+  constructor(path, flags, mode, fd) {
     super(path, flags, mode);
     this.#fd = fd;
-    this.#realPath = realPath;
   }
 
   readSync(buffer, offset, length, position) {
@@ -175,12 +193,72 @@ class RealFileHandle extends VirtualFileHandle {
 
   writeFileSync(data, options) {
     this.#checkClosed('write');
-    fs.writeFileSync(this.#realPath, data, options);
+    fs.writeFileSync(this.#fd, data, options);
+  }
+
+  // Writes the whole buffer at the descriptor's current position, at most
+  // kWriteFileMaxChunkSize per call, the way writeFileHandle() does.
+  async #writeAll(buffer, signal) {
+    let written = 0;
+    while (written < buffer.byteLength) {
+      checkAborted(signal);
+      const { bytesWritten } = await this.write(
+        buffer,
+        written,
+        MathMin(kWriteFileMaxChunkSize, buffer.byteLength - written),
+        null);
+      written += bytesWritten;
+    }
+  }
+
+  #fsync() {
+    this.#checkClosed('fsync');
+    return new Promise((resolve, reject) => {
+      fs.fsync(this.#fd, (err) => {
+        if (err) reject(err);
+        else resolve();
+      });
+    });
   }
 
   async writeFile(data, options) {
     this.#checkClosed('write');
-    return fs.promises.writeFile(this.#realPath, data, options);
+    if (!isCustomIterable(data)) {
+      return new Promise((resolve, reject) => {
+        fs.writeFile(this.#fd, data, options, (err) => {
+          if (err) reject(err);
+          else resolve();
+        });
+      });
+    }
+
+    // The chunks are written one at a time through the descriptor, the way
+    // writeFileHandle() does.
+    // `flush` is not part of that: `filehandle.writeFile()` ignores it, but
+    // reopening the path used to fsync once at the end, so that is kept here
+    // instead of being multiplied by the number of chunks.
+    const opts = getOptions(options, {
+      encoding: 'utf8',
+      mode: 0o666,
+      flush: false,
+    });
+    const flush = opts.flush ?? false;
+    validateBoolean(flush, 'options.flush');
+    parseFileMode(opts.mode, 'mode', 0o666);
+    // An already aborted signal must not end up waiting on a source that
+    // never yields, so it is read before the first next().
+    checkAborted(opts.signal);
+
+    const encoding = opts.encoding || 'utf8';
+    for await (const chunk of data) {
+      await this.#writeAll(
+        isArrayBufferView(chunk) ? chunk : Buffer.from(chunk, encoding),
+        opts.signal);
+      // An abort that arrived while the write was in flight must surface
+      // before the source is asked for another chunk.
+      checkAborted(opts.signal);
+    }
+    if (flush) await this.#fsync();
   }
 
   statSync(options) {
@@ -335,15 +413,15 @@ class RealFSProvider extends VirtualProvider {
   openSync(vfsPath, flags, mode) {
     const realPath = this.#resolvePath(vfsPath);
     const fd = fs.openSync(realPath, flags, mode);
-    return new RealFileHandle(vfsPath, flags, mode ?? 0o644, fd, realPath);
+    return new RealFileHandle(vfsPath, flags, mode ?? 0o644, fd);
   }
 
   async open(vfsPath, flags, mode) {
     const realPath = this.#resolvePath(vfsPath);
     return new Promise((resolve, reject) => {
       fs.open(realPath, flags, mode, (err, fd) => {
         if (err) reject(err);
-        else resolve(new RealFileHandle(vfsPath, flags, mode ?? 0o644, fd, realPath));
+        else resolve(new RealFileHandle(vfsPath, flags, mode ?? 0o644, fd));
       });
     });
   }

test/parallel/test-vfs-real-provider-handle.js

@@ -34,12 +34,134 @@ const myVfs = vfs.create(new vfs.RealFSProvider(root));
     assert.strictEqual(handle.statSync().isFile(), true);
     assert.strictEqual(handle.readFileSync('utf8'), 'zzllo world');
 
+    // Like `filehandle.writeFile()`, this writes from the handle's current
+    // position rather than replacing the file, so a shorter write over an
+    // "r+" handle leaves the tail of the old content in place.
     handle.writeFileSync('replaced');
-    assert.strictEqual(handle.readFileSync('utf8'), 'replaced');
+    assert.strictEqual(handle.readFileSync('utf8'), 'replacedrld');
 
     myVfs.closeSync(fd);
   }
 
+  // ===== writeFile goes through the file description, not the path =====
+  {
+    fs.writeFileSync(path.join(root, 'renamed-away.txt'), 'aaaaaa');
+    const handle = await myVfs.provider.open('/renamed-away.txt', 'r+');
+    fs.renameSync(path.join(root, 'renamed-away.txt'),
+                  path.join(root, 'renamed-to.txt'));
+
+    handle.writeFileSync('bb');
+    await handle.writeFile('cc');
+    await handle.close();
+
+    assert.strictEqual(
+      fs.readFileSync(path.join(root, 'renamed-to.txt'), 'utf8'), 'bbccaa');
+    assert.strictEqual(fs.existsSync(path.join(root, 'renamed-away.txt')),
+                       false);
+  }
+
+  // ===== writeFile takes the iterables filehandle.writeFile() takes =====
+  {
+    const handle = await myVfs.provider.open('/iterable.txt', 'w');
+    await handle.writeFile(['one ', 'two ']);
+    await handle.writeFile(async function* () {
+      yield 'three ';
+      yield Buffer.from('four');
+    }());
+    // A chunk that is not a view is converted the way writeFileHandle() does.
+    await handle.writeFile([[32, 65], Uint8Array.of(66).buffer]);
+    await handle.close();
+
+    assert.strictEqual(
+      fs.readFileSync(path.join(root, 'iterable.txt'), 'utf8'),
+      'one two three four AB');
+  }
+
+  // ===== options are validated before the source is consumed =====
+  {
+    const handle = await myVfs.provider.open('/opts.txt', 'w');
+
+    // The signal is read before the first next(), so a source that never
+    // yields cannot leave the write pending.
+    const neverYields = {
+      [Symbol.asyncIterator]: () => ({ next: () => new Promise(() => {}) }),
+    };
+    await assert.rejects(
+      handle.writeFile(neverYields, { signal: AbortSignal.abort() }),
+      { name: 'AbortError' });
+
+    // The rest of `options` is validated there too, so a bad value is
+    // reported instead of waiting on a source that never produces.
+    await assert.rejects(handle.writeFile(neverYields, { mode: 'invalid' }),
+                         { code: 'ERR_INVALID_ARG_VALUE' });
+
+    await handle.close();
+  }
+
+  // ===== flush costs one fsync per call, not one per chunk =====
+  {
+    const originalFsync = fs.fsync;
+    let fsyncs = 0;
+    fs.fsync = function fsync(...args) {
+      fsyncs++;
+      return originalFsync.apply(this, args);
+    };
+
+    try {
+      const handle = await myVfs.provider.open('/flushed.txt', 'w');
+      await handle.writeFile(['a', 'b', 'c'], { flush: true });
+      await handle.close();
+      assert.strictEqual(fsyncs, 1);
+      assert.strictEqual(
+        fs.readFileSync(path.join(root, 'flushed.txt'), 'utf8'), 'abc');
+
+    } finally {
+      fs.fsync = originalFsync;
+    }
+  }
+
+  // ===== an abort landing during a write stops the source =====
+  {
+    const handle = await myVfs.provider.open('/abort-mid.txt', 'w');
+    const ac = new AbortController();
+    const originalWrite = fs.write;
+    // Abort as the write settles, which is the window the post-write check
+    // covers. Without it a source of one chunk resolves successfully.
+    fs.write = function write(fd, buf, off, len, pos, callback) {
+      return originalWrite.call(this, fd, buf, off, len, pos, (err, n) => {
+        ac.abort();
+        callback(err, n);
+      });
+    };
+
+    let pulled = 0;
+    try {
+      await assert.rejects(handle.writeFile(async function* () {
+        pulled++;
+        yield 'first';
+        pulled++;
+        yield 'second';
+      }(), { signal: ac.signal }), { name: 'AbortError' });
+    } finally {
+      fs.write = originalWrite;
+      await handle.close();
+    }
+    assert.strictEqual(pulled, 1);  // The source was not asked for more
+  }
+
+  // ===== writeFile on a handle that was not opened for writing =====
+  {
+    fs.writeFileSync(path.join(root, 'ronly.txt'), 'untouched');
+    const handle = await myVfs.provider.open('/ronly.txt', 'r');
+
+    assert.throws(() => handle.writeFileSync('x'), { code: 'EBADF' });
+    await assert.rejects(handle.writeFile('x'), { code: 'EBADF' });
+    await handle.close();
+
+    assert.strictEqual(
+      fs.readFileSync(path.join(root, 'ronly.txt'), 'utf8'), 'untouched');
+  }
+
   // ===== Async read/write/stat/truncate via provider.open =====
   {
     await myVfs.promises.writeFile('/h2.txt', 'abcdef');
@@ -64,10 +186,13 @@ const myVfs = vfs.create(new vfs.RealFSProvider(root));
     assert.ok(handle.readFileSync().length > 0);
     assert.ok((await handle.readFile()).length > 0);
 
+    // Each write starts where the previous one left the handle, so the
+    // second call appends rather than replacing what the first one wrote.
     handle.writeFileSync('OVERWRITTEN');
     assert.strictEqual(handle.readFileSync('utf8'), 'OVERWRITTEN');
     await handle.writeFile('async-overwrite');
-    assert.strictEqual(await handle.readFile('utf8'), 'async-overwrite');
+    assert.strictEqual(await handle.readFile('utf8'),
+                       'OVERWRITTENasync-overwrite');
 
     handle.truncateSync(3);
     await handle.truncate(2);