fix(ext/io): cancel pending FileResource reads on close (#34544)

## Summary

A `ReadableStream` backed by a file resource (`Deno.FsFile.readable`)
could not be cancelled while a read was in flight. Reads on pipes /
FIFOs may block in the kernel indefinitely, so the in-flight `op_read`
kept the runtime alive long after JS had finished, preventing the
process from exiting.

This PR adds a `CancelHandle` to `FileResource`, wraps `read` /
`read_byob` with `try_or_cancel`, and triggers the handle from
`Resource::close()`. The op completes promptly on the JS side when the
stream is cancelled; the spawned blocking thread may still complete its
own kernel read (we cannot interrupt `spawn_blocking`), but its result
is discarded and the runtime no longer waits on it.

This mirrors the pattern already used by `ChildStdoutResource`,
`TcpStreamResource`, etc.

Fixes #21186.

## Reproducer

Before the fix, this would print `OK` and then hang indefinitely:

```ts
// open both ends so the open syscall doesn't block
const file = await Deno.open("/path/to/fifo", { read: true, write: true });
const r = file.readable.getReader();
const reading = r.read();
await new Promise((res) => setTimeout(res, 50));
await r.cancel();
await reading; // resolves with {done:true}
console.log("OK"); // runtime then hangs on the orphaned op_read
```

After the fix, the same script prints `OK` and exits cleanly.

## Test plan

- New unit test `readableStreamCancelOnFifo` in
`tests/unit/files_test.ts` (Unix only): spawns a subprocess that
performs the above, with a 10s watchdog. Verified it fails on `main` and
passes with this PR.
- Existing `readableStream` / `readableStreamTextEncoderPipe` tests
still pass.
- `cargo clippy -p deno_io` clean.

Closes denoland/orchid#288

---------

Co-authored-by: divybot <divybot@users.noreply.github.com>
Co-authored-by: Divy Srivastava <me@littledivy.com>

Author: divybot

ext/io/fs.rs

@@ -12,7 +12,11 @@ use std::time::UNIX_EPOCH;
 
 use deno_core::BufMutView;
 use deno_core::BufView;
+use deno_core::CancelHandle;
+use deno_core::CancelTryFuture;
+use deno_core::Canceled;
 use deno_core::OpState;
+use deno_core::RcRef;
 use deno_core::ResourceHandleFd;
 use deno_core::ResourceId;
 use deno_core::error::ResourceError;
@@ -98,6 +102,12 @@ impl From<JoinError> for FsError {
   }
 }
 
+impl From<Canceled> for FsError {
+  fn from(err: Canceled) -> Self {
+    Self::Io(err.into())
+  }
+}
+
 pub type FsResult<T> = Result<T, FsError>;
 
 pub struct FsStat {
@@ -345,11 +355,24 @@ pub trait File {
 pub struct FileResource {
   name: String,
   file: Rc<dyn File>,
+  /// Cancels pending read operations when the resource is closed.
+  /// Used so streams backed by a file (especially pipes / FIFOs whose
+  /// reads can block indefinitely) can be cancelled — see
+  /// <https://github.com/denoland/deno/issues/21186>.
+  cancel_handle: CancelHandle,
 }
 
 impl FileResource {
   pub fn new(file: Rc<dyn File>, name: String) -> Self {
-    Self { name, file }
+    Self {
+      name,
+      file,
+      cancel_handle: Default::default(),
+    }
+  }
+
+  fn cancel_handle(self: &Rc<Self>) -> RcRef<CancelHandle> {
+    RcRef::map(self, |r| &r.cancel_handle)
   }
 
   fn with_resource<F, R>(
@@ -397,11 +420,13 @@ impl deno_core::Resource for FileResource {
   }
 
   fn read(self: Rc<Self>, limit: usize) -> deno_core::AsyncResult<BufView> {
+    let cancel_handle = self.cancel_handle();
     Box::pin(async move {
       self
         .file
         .clone()
         .read(limit)
+        .try_or_cancel(cancel_handle)
         .await
         .map_err(JsErrorBox::from_err)
     })
@@ -411,11 +436,13 @@ impl deno_core::Resource for FileResource {
     self: Rc<Self>,
     buf: BufMutView,
   ) -> deno_core::AsyncResult<(usize, BufMutView)> {
+    let cancel_handle = self.cancel_handle();
     Box::pin(async move {
       self
         .file
         .clone()
         .read_byob(buf)
+        .try_or_cancel(cancel_handle)
         .await
         .map_err(JsErrorBox::from_err)
     })
@@ -468,4 +495,12 @@ impl deno_core::Resource for FileResource {
   fn backing_fd(self: Rc<Self>) -> Option<ResourceHandleFd> {
     self.file.clone().backing_fd()
   }
+
+  fn close(self: Rc<Self>) {
+    // Cancel any pending read operations. Reads on pipes / FIFOs can
+    // block indefinitely; without this, a `ReadableStream` backed by
+    // such a file could never have its `cancel()` promise resolve while
+    // a read is outstanding.
+    self.cancel_handle.cancel();
+  }
 }

tests/unit/files_test.ts

@@ -640,6 +640,61 @@ Deno.test(
   },
 );
 
+Deno.test(
+  {
+    permissions: { read: true, write: true, run: true },
+    ignore: Deno.build.os === "windows",
+  },
+  async function readableStreamCancelOnFifo() {
+    // Regression test for https://github.com/denoland/deno/issues/21186
+    // — a `read` on a FIFO can block indefinitely waiting for data, and
+    // previously cancelling the wrapping `ReadableStream` would not
+    // unblock the in-flight `op_read`. As a result, the runtime would
+    // hang on exit even after JS had finished. We verify the fix by
+    // running a subprocess that opens a FIFO, starts a read, cancels
+    // it, and is expected to exit promptly. Without the fix the
+    // subprocess would hang until killed.
+    const tempDir = await Deno.makeTempDir();
+    const fifo = `${tempDir}/fifo`;
+    try {
+      const mkfifo = new Deno.Command("mkfifo", { args: [fifo] }).outputSync();
+      if (mkfifo.code !== 0) {
+        throw new Error(
+          `mkfifo failed: ${new TextDecoder().decode(mkfifo.stderr)}`,
+        );
+      }
+      const script = `
+        const fifo = ${JSON.stringify(fifo)};
+        // O_RDWR avoids the POSIX FIFO open-blocks-on-peer semantics.
+        const file = await Deno.open(fifo, { read: true, write: true });
+        const r = file.readable.getReader();
+        const reading = r.read();
+        await new Promise((res) => setTimeout(res, 50));
+        await r.cancel();
+        await reading;
+      `;
+      const out = await new Deno.Command(Deno.execPath(), {
+        args: ["eval", script],
+        stdout: "null",
+        stderr: "piped",
+        signal: AbortSignal.timeout(10_000),
+      }).output();
+      if (out.signal !== null) {
+        throw new Error(
+          "subprocess hung after stream cancellation — read was not cancelled",
+        );
+      }
+      assertEquals(
+        out.code,
+        0,
+        `subprocess failed: ${new TextDecoder().decode(out.stderr)}`,
+      );
+    } finally {
+      await Deno.remove(tempDir, { recursive: true });
+    }
+  },
+);
+
 Deno.test(
   { permissions: { read: true, write: true } },
   async function writableStream() {

tests/wpt/runner/runner.ts

@@ -179,7 +179,12 @@ async function runSingleTestInner(
     interval = setInterval(() => {
       const passedTime = performance.now() - start;
       if (passedTime > timeout) {
-        proc.kill("SIGINT");
+        try {
+          proc.kill("SIGINT");
+        } catch {
+          // Race: proc may have terminated between the for-await loop
+          // draining stderr and `clearInterval` running in `finally`.
+        }
       }
     }, 1000);
     for await (const line of lines as AsyncIterable<string>) {