fs: add signal option to filehandle.readableWebStream()

Author: LiviaMedeiros

doc/api/fs.md

@@ -481,6 +481,9 @@ number of bytes read is zero.
 <!-- YAML
 added: v17.0.0
 changes:
+  - version: REPLACEME
+    pr-url: https://github.com/nodejs/node/pull/58725
+    description: Added the `signal` option.
   - version: v24.2.0
     pr-url: https://github.com/nodejs/node/pull/58548
     description: Added the `autoClose` option.
@@ -502,6 +505,7 @@ changes:
 * `options` {Object}
   * `autoClose` {boolean} When true, causes the {FileHandle} to be closed when the
     stream is closed. **Default:** `false`
+  * `signal` {AbortSignal|undefined} allows aborting the stream. **Default:** `undefined`
 * Returns: {ReadableStream}
 
 Returns a byte-oriented `ReadableStream` that may be used to read the file's

lib/internal/fs/promises.js

@@ -278,7 +278,11 @@ class FileHandle extends EventEmitter {
   /**
    * @typedef {import('../webstreams/readablestream').ReadableStream
    * } ReadableStream
-   * @param {{ type?: 'bytes', autoClose?: boolean }} [options]
+   * @param {{
+      type?: 'bytes';
+      autoClose?: boolean;
+      signal?: AbortSignal;
+      }} [options]
    * @returns {ReadableStream}
    */
   readableWebStream(options = kEmptyObject) {
@@ -294,9 +298,18 @@ class FileHandle extends EventEmitter {
     const {
       type = 'bytes',
       autoClose = false,
+      signal,
     } = options;
 
     validateBoolean(autoClose, 'options.autoClose');
+    validateAbortSignal(signal, 'options.signal');
+
+    // Can't use checkAborted() because we may need to close filehandle first
+    if (signal?.aborted) {
+      if (autoClose) this.close();
+      throw new AbortError(undefined, { cause: signal.reason });
+    }
+    let signalListenerCleanup = null;
 
     if (type !== 'bytes') {
       process.emitWarning(
@@ -308,6 +321,7 @@ class FileHandle extends EventEmitter {
 
     const readFn = FunctionPrototypeBind(this.read, this);
     const ondone = async () => {
+      signalListenerCleanup?.();
       this[kUnref]();
       if (autoClose) await this.close();
     };
@@ -317,6 +331,21 @@ class FileHandle extends EventEmitter {
       type: 'bytes',
       autoAllocateChunkSize: 16384,
 
+      start: (controller) => {
+        if (signal) {
+          const onAbort = async () => {
+            this.off('close', cancelOnClose);
+            controller.error(new AbortError(undefined, { cause: signal.reason }));
+            await ondone();
+          };
+
+          signal.addEventListener('abort', onAbort, { __proto__: null, once: true });
+          signalListenerCleanup = () => {
+            signal.removeEventListener('abort', onAbort);
+          };
+        }
+      },
+
       async pull(controller) {
         const view = controller.byobRequest.view;
         const { bytesRead } = await readFn(view, view.byteOffset, view.byteLength);
@@ -339,9 +368,11 @@ class FileHandle extends EventEmitter {
       readableStreamCancel,
     } = require('internal/webstreams/readablestream');
     this[kRef]();
-    this.once('close', () => {
+
+    const cancelOnClose = () => {
       readableStreamCancel(readable);
-    });
+    };
+    this.once('close', cancelOnClose);
 
     return readable;
   }

test/parallel/test-filehandle-readablewebstream-signal.mjs

@@ -0,0 +1,131 @@
+import { skip } from '../common/index.mjs';
+import { open, access, constants } from 'node:fs/promises';
+import assert from 'node:assert';
+
+// Test that signal option in filehandle.readableWebStream() works
+
+const shortFile = new URL(import.meta.url);
+
+{
+  await using fh = await open(shortFile);
+  const cause = new Error('am abort reason');
+  const controller = new AbortController();
+  const { signal } = controller;
+  controller.abort(cause);
+  assert.throws(() => fh.readableWebStream({ signal, autoClose: false }), {
+    code: 'ABORT_ERR',
+    cause,
+  });
+
+  // Filehandle must be still open
+  await fh.read();
+}
+
+{
+  await using fh = await open(shortFile);
+  const cause = new Error('am abort reason');
+  const controller = new AbortController();
+  const { signal } = controller;
+  controller.abort(cause);
+  assert.throws(() => fh.readableWebStream({ signal, autoClose: true }), {
+    code: 'ABORT_ERR',
+    cause,
+  });
+
+  // Filehandle must be closed after abort
+  await assert.rejects(() => fh.read(), {
+    code: 'EBADF',
+  });
+}
+
+{
+  await using fh = await open(shortFile);
+  const cause = new Error('am abort reason');
+  const controller = new AbortController();
+  const { signal } = controller;
+  const stream = fh.readableWebStream({ signal, autoClose: false });
+  const reader = stream.getReader();
+  controller.abort(cause);
+  await assert.rejects(() => reader.read(), {
+    code: 'ABORT_ERR',
+    cause,
+  });
+
+  // Filehandle must be still open
+  await fh.read();
+}
+
+{
+  await using fh = await open(shortFile);
+  const cause = new Error('am abort reason');
+  const controller = new AbortController();
+  const { signal } = controller;
+  const stream = fh.readableWebStream({ signal, autoClose: true });
+  const reader = stream.getReader();
+  controller.abort(cause);
+  await assert.rejects(() => reader.read(), {
+    code: 'ABORT_ERR',
+    cause,
+  });
+
+  // Filehandle must be closed after abort
+  await assert.rejects(() => fh.read(), {
+    code: 'EBADF',
+  });
+}
+
+const longFile = new URL('file:///dev/zero');
+
+try {
+  await access(longFile, constants.R_OK);
+} catch {
+  skip('Can not perform long test');
+}
+
+{
+  await using fh = await open(longFile);
+  const cause = new Error('am abort reason');
+  const controller = new AbortController();
+  const { signal } = controller;
+  const stream = fh.readableWebStream({ signal, autoClose: false });
+  const reader = stream.getReader();
+  setTimeout(() => controller.abort(cause), 100);
+  await assert.rejects(async () => {
+    while (true) {
+      await new Promise((resolve) => setTimeout(resolve, 5));
+      const { done } = await reader.read();
+      assert.ok(done === false, 'we exhausted /dev/zero');
+    }
+  }, {
+    code: 'ABORT_ERR',
+    cause,
+  });
+
+  // Filehandle must be still open
+  await fh.read();
+}
+
+{
+  await using fh = await open(longFile);
+  const cause = new Error('am abort reason');
+  const controller = new AbortController();
+  const { signal } = controller;
+  const stream = fh.readableWebStream({ signal, autoClose: true });
+  const reader = stream.getReader();
+  setTimeout(() => controller.abort(cause), 100);
+  await assert.rejects(async () => {
+    while (true) {
+      await new Promise((resolve) => setTimeout(resolve, 5));
+      const { done } = await reader.read();
+      assert.ok(done === false, 'we exhausted /dev/zero');
+    }
+  }, {
+    code: 'ABORT_ERR',
+    cause,
+  });
+
+  // Filehandle must be closed after abort
+  await assert.rejects(() => fh.read(), {
+    code: 'EBADF',
+  });
+}