fs, stream: add initial Symbol.dispose and Symbol.asyncDispose support

doc/api/fs.md

@@ -817,6 +817,16 @@ On Linux, positional writes don't work when the file is opened in append mode.
 The kernel ignores the position argument and always appends the data to
 the end of the file.
 
+#### `filehandle[Symbol.asyncDispose]()`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+An alias for `filehandle.close()`.
+
 ### `fsPromises.access(path[, mode])`
 
 <!-- YAML

doc/api/stream.md

@@ -1904,6 +1904,17 @@ option. In the code example above, data will be in a single chunk if the file
 has less then 64 KiB of data because no `highWaterMark` option is provided to
 [`fs.createReadStream()`][].
 
+##### `readable[Symbol.asyncDispose]()`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+Calls [`readable.destroy()`][readable-destroy] with an `AbortError` and returns
+a promise that fulfills when the stream is finished.
+
 ##### `readable.compose(stream[, options])`
 
 <!-- YAML

lib/internal/fs/promises.js

@@ -14,6 +14,7 @@ const {
   SafeArrayIterator,
   SafePromisePrototypeFinally,
   Symbol,
+  SymbolAsyncDispose,
   Uint8Array,
   FunctionPrototypeBind,
 } = primordials;
@@ -246,6 +247,10 @@ class FileHandle extends EventEmitterMixin(JSTransferable) {
     return this[kClosePromise];
   };
 
+  async [SymbolAsyncDispose]() {
+    return this.close();
+  }
+
   /**
    * @typedef {import('../webstreams/readablestream').ReadableStream
    * } ReadableStream

lib/internal/per_context/primordials.js

@@ -228,6 +228,12 @@ function copyPrototype(src, dest, prefix) {
   copyPrototype(original.prototype, primordials, `${name}Prototype`);
 });
 
+// Define Symbol.Dispose and Symbol.AsyncDispose
+// Until these are defined by the environment.
+// TODO(MoLow): Remove this polyfill once Symbol.dispose and Symbol.asyncDispose are available in V8.
+primordials.SymbolDispose ??= primordials.SymbolFor('nodejs.dispose');
+primordials.SymbolAsyncDispose ??= primordials.SymbolFor('nodejs.asyncDispose');
+
 // Create copies of intrinsic objects that require a valid `this` to call
 // static methods.
 // Refs: https://www.ecma-international.org/ecma-262/#sec-promise.all

lib/internal/process/pre_execution.js

@@ -8,6 +8,9 @@ const {
   ObjectGetOwnPropertyDescriptor,
   SafeMap,
   StringPrototypeStartsWith,
+  Symbol,
+  SymbolDispose,
+  SymbolAsyncDispose,
   globalThis,
 } = primordials;
 
@@ -82,6 +85,8 @@ function prepareExecution(options) {
 
   require('internal/dns/utils').initializeDns();
 
+  setupSymbolDisposePolyfill();
+
   if (isMainThread) {
     assert(internalBinding('worker').isMainThread);
     // Worker threads will get the manifest in the message handler.
@@ -119,6 +124,14 @@ function prepareExecution(options) {
   }
 }
 
+function setupSymbolDisposePolyfill() {
+  // TODO(MoLow): Remove this polyfill once Symbol.dispose and Symbol.asyncDispose are available in V8.
+  // eslint-disable-next-line node-core/prefer-primordials
+  Symbol.dispose ??= SymbolDispose;
+  // eslint-disable-next-line node-core/prefer-primordials
+  Symbol.asyncDispose ??= SymbolAsyncDispose;
+}
+
 function setupUserModules(isLoaderWorker = false) {
   initializeCJSLoader();
   initializeESMLoader(isLoaderWorker);

lib/internal/streams/readable.js

@@ -31,6 +31,7 @@ const {
   ObjectSetPrototypeOf,
   Promise,
   SafeSet,
+  SymbolAsyncDispose,
   SymbolAsyncIterator,
   Symbol,
 } = primordials;
@@ -67,6 +68,7 @@ const {
     ERR_STREAM_UNSHIFT_AFTER_END_EVENT,
     ERR_UNKNOWN_ENCODING,
   },
+  AbortError,
 } = require('internal/errors');
 const { validateObject } = require('internal/validators');
 
@@ -234,6 +236,15 @@ Readable.prototype[EE.captureRejectionSymbol] = function(err) {
   this.destroy(err);
 };
 
+Readable.prototype[SymbolAsyncDispose] = function() {
+  let error;
+  if (!this.destroyed) {
+    error = this.readableEnded ? null : new AbortError();
+    this.destroy(error);
+  }
+  return new Promise((resolve, reject) => eos(this, (err) => (err && err !== error ? reject(err) : resolve(null))));
+};
+
 // Manually shove something into the read() buffer.
 // This returns true if the highWaterMark has not been hit yet,
 // similar to how Writable.write() returns true if you should

test/parallel/test-fs-promises-file-handle-dispose.js

@@ -0,0 +1,12 @@
+'use strict';
+
+const common = require('../common');
+const { promises: fs } = require('fs');
+
+async function doOpen() {
+  const fh = await fs.open(__filename);
+  fh.on('close', common.mustCall());
+  await fh[Symbol.asyncDispose]();
+}
+
+doOpen().then(common.mustCall());

test/parallel/test-stream-readable-dispose.js

@@ -0,0 +1,23 @@
+'use strict';
+
+const common = require('../common');
+const { Readable } = require('stream');
+const assert = require('assert');
+
+{
+  const read = new Readable({
+    read() {}
+  });
+  read.resume();
+
+  read.on('end', common.mustNotCall('no end event'));
+  read.on('close', common.mustCall());
+  read.on('error', common.mustCall((err) => {
+    assert.strictEqual(err.name, 'AbortError');
+  }));
+
+  read[Symbol.asyncDispose]().then(common.mustCall(() => {
+    assert.strictEqual(read.errored.name, 'AbortError');
+    assert.strictEqual(read.destroyed, true);
+  }));
+}

typings/primordials.d.ts

@@ -480,6 +480,8 @@ declare namespace primordials {
   export const SymbolFor: typeof Symbol.for
   export const SymbolKeyFor: typeof Symbol.keyFor
   export const SymbolAsyncIterator: typeof Symbol.asyncIterator
+  export const SymbolDispose: typeof Symbol // TODO(MoLow): use typeof Symbol.dispose when it's available
+  export const SymbolAsyncDispose: typeof Symbol // TODO(MoLow): use typeof Symbol.asyncDispose when it's available
   export const SymbolHasInstance: typeof Symbol.hasInstance
   export const SymbolIsConcatSpreadable: typeof Symbol.isConcatSpreadable
   export const SymbolIterator: typeof Symbol.iterator