fs, stream: add initial Symbol.dispose and Symbol.asyncDispose su…

…pport

Co-authored-by: Benjamin Gruenbaum <benjamingr@gmail.com>

doc/api/fs.md

@@ -817,6 +817,14 @@ 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.dispose]()`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+an alias for `filehandle.close()`.
+
 ### `fsPromises.access(path[, mode])`
 
 <!-- YAML

doc/api/stream.md

@@ -1904,6 +1904,14 @@ 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.dispose]()`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+an alias for [`readable.destroy()`][readable-destroy] with an `AbortError`.
+
 ##### `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,11 @@ function copyPrototype(src, dest, prefix) {
   copyPrototype(original.prototype, primordials, `${name}Prototype`);
 });
 
+// Define Symbol.Disposed and Symbol.AsyncDispose
+// Until these are defined by the environment.
+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();
 
+  setupSymbolDispoasePolyfill();
+
   if (isMainThread) {
     assert(internalBinding('worker').isMainThread);
     // Worker threads will get the manifest in the message handler.
@@ -119,6 +124,12 @@ function prepareExecution(options) {
   }
 }
 
+function setupSymbolDispoasePolyfill() {
+  Symbol.dispose ??= SymbolDispose;
+  Symbol.asyncDispose ??= SymbolAsyncDispose;
+}
+
+
 function setupUserModules(isLoaderWorker = false) {
   initializeCJSLoader();
   initializeESMLoader(isLoaderWorker);

lib/internal/streams/readable.js

@@ -31,6 +31,7 @@ const {
   ObjectSetPrototypeOf,
   Promise,
   SafeSet,
+  SymbolDispose,
   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,10 @@ Readable.prototype[EE.captureRejectionSymbol] = function(err) {
   this.destroy(err);
 };
 
+Readable.prototype[SymbolDispose] = function() {
+  this.destroy(new AbortError());
+};
+
 // 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,22 @@
+'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.dispose]();
+  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
+  export const SymbolAsyncDispose: typeof Symbol
   export const SymbolHasInstance: typeof Symbol.hasInstance
   export const SymbolIsConcatSpreadable: typeof Symbol.isConcatSpreadable
   export const SymbolIterator: typeof Symbol.iterator