fix(ops): Always close cancel handles for read_async/write_async (#17736)

Fixes https://github.com/denoland/deno/issues/17734

cli/tests/unit/read_file_test.ts

@@ -140,6 +140,17 @@ Deno.test(
   },
 );
 
+// Test that AbortController's cancel handle is cleaned-up correctly, and do not leak resources.
+Deno.test(
+  { permissions: { read: true } },
+  async function readFileWithAbortSignalNotCalled() {
+    const ac = new AbortController();
+    await Deno.readFile("cli/tests/testdata/assets/fixture.json", {
+      signal: ac.signal,
+    });
+  },
+);
+
 Deno.test(
   { permissions: { read: true }, ignore: Deno.build.os !== "linux" },
   async function readFileProcFs() {

cli/tests/unit/read_text_file_test.ts

@@ -95,7 +95,7 @@ Deno.test(
     queueMicrotask(() => ac.abort());
     const error = await assertRejects(
       async () => {
-        await Deno.readFile("cli/tests/testdata/assets/fixture.json", {
+        await Deno.readTextFile("cli/tests/testdata/assets/fixture.json", {
           signal: ac.signal,
         });
       },
@@ -113,7 +113,7 @@ Deno.test(
     queueMicrotask(() => ac.abort(abortReason));
     const error = await assertRejects(
       async () => {
-        await Deno.readFile("cli/tests/testdata/assets/fixture.json", {
+        await Deno.readTextFile("cli/tests/testdata/assets/fixture.json", {
           signal: ac.signal,
         });
       },
@@ -128,7 +128,7 @@ Deno.test(
     const ac = new AbortController();
     queueMicrotask(() => ac.abort("Some string"));
     try {
-      await Deno.readFile("cli/tests/testdata/assets/fixture.json", {
+      await Deno.readTextFile("cli/tests/testdata/assets/fixture.json", {
         signal: ac.signal,
       });
       unreachable();
@@ -138,6 +138,17 @@ Deno.test(
   },
 );
 
+// Test that AbortController's cancel handle is cleaned-up correctly, and do not leak resources.
+Deno.test(
+  { permissions: { read: true } },
+  async function readTextFileWithAbortSignalNotCalled() {
+    const ac = new AbortController();
+    await Deno.readTextFile("cli/tests/testdata/assets/fixture.json", {
+      signal: ac.signal,
+    });
+  },
+);
+
 Deno.test(
   { permissions: { read: true }, ignore: Deno.build.os !== "linux" },
   async function readTextFileProcFs() {

cli/tests/unit/write_file_test.ts

@@ -379,6 +379,22 @@ Deno.test(
   },
 );
 
+// Test that AbortController's cancel handle is cleaned-up correctly, and do not leak resources.
+Deno.test(
+  { permissions: { read: true, write: true } },
+  async function writeFileWithAbortSignalNotCalled() {
+    const ac = new AbortController();
+    const enc = new TextEncoder();
+    const data = enc.encode("Hello");
+    const filename = Deno.makeTempDirSync() + "/test.txt";
+    await Deno.writeFile(filename, data, { signal: ac.signal });
+    const dataRead = Deno.readFileSync(filename);
+    const dec = new TextDecoder("utf-8");
+    const actual = dec.decode(dataRead);
+    assertEquals(actual, "Hello");
+  },
+);
+
 function assertNotExists(filename: string | URL) {
   if (pathExists(filename)) {
     throw new Error(`The file ${filename} exists.`);

runtime/ops/fs.rs

@@ -267,14 +267,6 @@ async fn op_write_file_async(
   data: ZeroCopyBuf,
   cancel_rid: Option<ResourceId>,
 ) -> Result<(), AnyError> {
-  let cancel_handle = match cancel_rid {
-    Some(cancel_rid) => state
-      .borrow_mut()
-      .resource_table
-      .get::<CancelHandle>(cancel_rid)
-      .ok(),
-    None => None,
-  };
   let (path, open_options) = open_helper(
     &mut state.borrow_mut(),
     &path,
@@ -282,15 +274,30 @@ async fn op_write_file_async(
     Some(&write_open_options(create, append, create_new)),
     "Deno.writeFile()",
   )?;
+
   let write_future = tokio::task::spawn_blocking(move || {
     write_file(&path, open_options, mode, data)
   });
+
+  let cancel_handle = cancel_rid.and_then(|rid| {
+    state
+      .borrow_mut()
+      .resource_table
+      .get::<CancelHandle>(rid)
+      .ok()
+  });
+
   if let Some(cancel_handle) = cancel_handle {
-    write_future.or_cancel(cancel_handle).await???;
-  } else {
-    write_future.await??;
+    let write_future_rv = write_future.or_cancel(cancel_handle).await;
+
+    if let Some(cancel_rid) = cancel_rid {
+      state.borrow_mut().resource_table.close(cancel_rid).ok();
+    };
+
+    return write_future_rv??;
   }
-  Ok(())
+
+  write_future.await?
 }
 
 fn write_file(
@@ -2046,20 +2053,31 @@ async fn op_readfile_async(
       .borrow_mut::<PermissionsContainer>()
       .check_read(path, "Deno.readFile()")?;
   }
-  let fut = tokio::task::spawn_blocking(move || {
+
+  let read_future = tokio::task::spawn_blocking(move || {
     let path = Path::new(&path);
     Ok(std::fs::read(path).map(ZeroCopyBuf::from)?)
   });
-  if let Some(cancel_rid) = cancel_rid {
-    let cancel_handle = state
+
+  let cancel_handle = cancel_rid.and_then(|rid| {
+    state
       .borrow_mut()
       .resource_table
-      .get::<CancelHandle>(cancel_rid);
-    if let Ok(cancel_handle) = cancel_handle {
-      return fut.or_cancel(cancel_handle).await??;
-    }
+      .get::<CancelHandle>(rid)
+      .ok()
+  });
+
+  if let Some(cancel_handle) = cancel_handle {
+    let read_future_rv = read_future.or_cancel(cancel_handle).await;
+
+    if let Some(cancel_rid) = cancel_rid {
+      state.borrow_mut().resource_table.close(cancel_rid).ok();
+    };
+
+    return read_future_rv??;
   }
-  fut.await?
+
+  read_future.await?
 }
 
 #[op]
@@ -2075,20 +2093,31 @@ async fn op_readfile_text_async(
       .borrow_mut::<PermissionsContainer>()
       .check_read(path, "Deno.readTextFile()")?;
   }
-  let fut = tokio::task::spawn_blocking(move || {
+
+  let read_future = tokio::task::spawn_blocking(move || {
     let path = Path::new(&path);
     Ok(string_from_utf8_lossy(std::fs::read(path)?))
   });
-  if let Some(cancel_rid) = cancel_rid {
-    let cancel_handle = state
+
+  let cancel_handle = cancel_rid.and_then(|rid| {
+    state
       .borrow_mut()
       .resource_table
-      .get::<CancelHandle>(cancel_rid);
-    if let Ok(cancel_handle) = cancel_handle {
-      return fut.or_cancel(cancel_handle).await??;
-    }
+      .get::<CancelHandle>(rid)
+      .ok()
+  });
+
+  if let Some(cancel_handle) = cancel_handle {
+    let read_future_rv = read_future.or_cancel(cancel_handle).await;
+
+    if let Some(cancel_rid) = cancel_rid {
+      state.borrow_mut().resource_table.close(cancel_rid).ok();
+    };
+
+    return read_future_rv??;
   }
-  fut.await?
+
+  read_future.await?
 }
 
 // Like String::from_utf8_lossy but operates on owned values