fix(ext/node): add fs.cp, fs.cpSync, promises.cp (#21745)

Fixes https://github.com/denoland/deno/issues/20803
Fixes https://github.com/denoland/deno/issues/21723

Performance: copying a 48GiB rust `target` folder (recursive)
| Platform  | `deno` | `node v21.5` | Improvement |
| -------- | ------- | ------- | ------- |
| macOS (APFS) |   3.1secs  |  127.99 secs |  **42x** |
| Windows | 18.3secs | 67.2secs |  **3.8x** |

Copying files with varying sizes:


![image](https://github.com/denoland/deno/assets/34997667/58932652-6f7a-47f5-8504-896dc9ab4ddc)

7 files changed, 278 insertions, 0 deletions

diff --git a/ext/fs/Cargo.toml b/ext/fs/Cargo.toml
index 2d5919c9d..4a10fac37 100644
--- a/ext/fs/Cargo.toml
+++ b/ext/fs/Cargo.toml
@@ -25,6 +25,7 @@ fs3.workspace = true
 libc.workspace = true
 log.workspace = true
 rand.workspace = true
+rayon = "1.8.0"
 serde.workspace = true
 tokio.workspace = true
 
diff --git a/ext/fs/interface.rs b/ext/fs/interface.rs
index e69e80c6b..8ffa61481 100644
--- a/ext/fs/interface.rs
+++ b/ext/fs/interface.rs
@@ -131,6 +131,9 @@ pub trait FileSystem: std::fmt::Debug + MaybeSend + MaybeSync {
     newpath: PathBuf,
   ) -> FsResult<()>;
 
+  fn cp_sync(&self, path: &Path, new_path: &Path) -> FsResult<()>;
+  async fn cp_async(&self, path: PathBuf, new_path: PathBuf) -> FsResult<()>;
+
   fn stat_sync(&self, path: &Path) -> FsResult<FsStat>;
   async fn stat_async(&self, path: PathBuf) -> FsResult<FsStat>;
 
diff --git a/ext/fs/std_fs.rs b/ext/fs/std_fs.rs
index d8e2f3085..d52879394 100644
--- a/ext/fs/std_fs.rs
+++ b/ext/fs/std_fs.rs
@@ -150,6 +150,13 @@ impl FileSystem for RealFs {
     spawn_blocking(move || copy_file(&from, &to)).await?
   }
 
+  fn cp_sync(&self, fro: &Path, to: &Path) -> FsResult<()> {
+    cp(fro, to)
+  }
+  async fn cp_async(&self, fro: PathBuf, to: PathBuf) -> FsResult<()> {
+    spawn_blocking(move || cp(&fro, &to)).await?
+  }
+
   fn stat_sync(&self, path: &Path) -> FsResult<FsStat> {
     stat(path).map(Into::into)
   }
@@ -469,6 +476,157 @@ fn copy_file(from: &Path, to: &Path) -> FsResult<()> {
   Ok(())
 }
 
+fn cp(from: &Path, to: &Path) -> FsResult<()> {
+  fn cp_(source_meta: fs::Metadata, from: &Path, to: &Path) -> FsResult<()> {
+    use rayon::prelude::IntoParallelIterator;
+    use rayon::prelude::ParallelIterator;
+
+    let ty = source_meta.file_type();
+    if ty.is_dir() {
+      #[allow(unused_mut)]
+      let mut builder = fs::DirBuilder::new();
+      #[cfg(unix)]
+      {
+        use std::os::unix::fs::DirBuilderExt;
+        use std::os::unix::fs::PermissionsExt;
+        builder.mode(fs::symlink_metadata(from)?.permissions().mode());
+      }
+      builder.create(to)?;
+
+      let mut entries: Vec<_> = fs::read_dir(from)?
+        .map(|res| res.map(|e| e.file_name()))
+        .collect::<Result<_, _>>()?;
+
+      entries.shrink_to_fit();
+      entries
+        .into_par_iter()
+        .map(|file_name| {
+          cp_(
+            fs::symlink_metadata(from.join(&file_name)).unwrap(),
+            &from.join(&file_name),
+            &to.join(&file_name),
+          )
+          .map_err(|err| {
+            io::Error::new(
+              err.kind(),
+              format!(
+                "failed to copy '{}' to '{}': {:?}",
+                from.join(&file_name).display(),
+                to.join(&file_name).display(),
+                err
+              ),
+            )
+          })
+        })
+        .collect::<Result<Vec<_>, _>>()?;
+
+      return Ok(());
+    } else if ty.is_symlink() {
+      let from = std::fs::read_link(from)?;
+
+      #[cfg(unix)]
+      std::os::unix::fs::symlink(from, to)?;
+      #[cfg(windows)]
+      std::os::windows::fs::symlink_file(from, to)?;
+
+      return Ok(());
+    }
+    #[cfg(unix)]
+    {
+      use std::os::unix::fs::FileTypeExt;
+      if ty.is_socket() {
+        return Err(
+          io::Error::new(
+            io::ErrorKind::InvalidInput,
+            "sockets cannot be copied",
+          )
+          .into(),
+        );
+      }
+    }
+    copy_file(from, to)
+  }
+
+  #[cfg(target_os = "macos")]
+  {
+    // Just clonefile()
+    use libc::clonefile;
+    use libc::unlink;
+    use std::ffi::CString;
+    use std::os::unix::ffi::OsStrExt;
+
+    let from_str = CString::new(from.as_os_str().as_bytes()).unwrap();
+    let to_str = CString::new(to.as_os_str().as_bytes()).unwrap();
+
+    // SAFETY: `from` and `to` are valid C strings.
+    unsafe {
+      // Try unlink. If it fails, we are going to try clonefile() anyway.
+      let _ = unlink(to_str.as_ptr());
+
+      if clonefile(from_str.as_ptr(), to_str.as_ptr(), 0) == 0 {
+        return Ok(());
+      }
+    }
+  }
+
+  let source_meta = fs::symlink_metadata(from)?;
+
+  #[inline]
+  fn is_identical(
+    source_meta: &fs::Metadata,
+    dest_meta: &fs::Metadata,
+  ) -> bool {
+    #[cfg(unix)]
+    {
+      use std::os::unix::fs::MetadataExt;
+      source_meta.ino() == dest_meta.ino()
+    }
+    #[cfg(windows)]
+    {
+      use std::os::windows::fs::MetadataExt;
+      // https://learn.microsoft.com/en-us/windows/win32/api/fileapi/ns-fileapi-by_handle_file_information
+      //
+      // The identifier (low and high parts) and the volume serial number uniquely identify a file on a single computer.
+      // To determine whether two open handles represent the same file, combine the identifier and the volume serial
+      // number for each file and compare them.
+      //
+      // Use this code once file_index() and volume_serial_number() is stabalized
+      // See: https://github.com/rust-lang/rust/issues/63010
+      //
+      // source_meta.file_index() == dest_meta.file_index()
+      //   && source_meta.volume_serial_number()
+      //     == dest_meta.volume_serial_number()
+      source_meta.last_write_time() == dest_meta.last_write_time()
+        && source_meta.creation_time() == dest_meta.creation_time()
+    }
+  }
+
+  match (fs::metadata(to), fs::symlink_metadata(to)) {
+    (Ok(m), _) if m.is_dir() => cp_(
+      source_meta,
+      from,
+      &to.join(from.file_name().ok_or_else(|| {
+        io::Error::new(
+          io::ErrorKind::InvalidInput,
+          "the source path is not a valid file",
+        )
+      })?),
+    )?,
+    (_, Ok(m)) if is_identical(&source_meta, &m) => {
+      return Err(
+        io::Error::new(
+          io::ErrorKind::InvalidInput,
+          "the source and destination are the same file",
+        )
+        .into(),
+      )
+    }
+    _ => cp_(source_meta, from, to)?,
+  }
+
+  Ok(())
+}
+
 #[cfg(not(windows))]
 fn stat(path: &Path) -> FsResult<FsStat> {
   let metadata = fs::metadata(path)?;
diff --git a/ext/node/lib.rs b/ext/node/lib.rs
index 2aac49754..de56285fd 100644
--- a/ext/node/lib.rs
+++ b/ext/node/lib.rs
@@ -63,6 +63,11 @@ pub trait NodePermissions {
     api_name: Option<&str>,
   ) -> Result<(), AnyError>;
   fn check_sys(&self, kind: &str, api_name: &str) -> Result<(), AnyError>;
+  fn check_write_with_api_name(
+    &self,
+    path: &Path,
+    api_name: Option<&str>,
+  ) -> Result<(), AnyError>;
 }
 
 pub(crate) struct AllowAllNodePermissions;
@@ -82,6 +87,13 @@ impl NodePermissions for AllowAllNodePermissions {
   ) -> Result<(), AnyError> {
     Ok(())
   }
+  fn check_write_with_api_name(
+    &self,
+    _path: &Path,
+    _api_name: Option<&str>,
+  ) -> Result<(), AnyError> {
+    Ok(())
+  }
   fn check_sys(&self, _kind: &str, _api_name: &str) -> Result<(), AnyError> {
     Ok(())
   }
@@ -238,6 +250,8 @@ deno_core::extension!(deno_node,
     ops::crypto::x509::op_node_x509_get_serial_number,
     ops::crypto::x509::op_node_x509_key_usage,
     ops::fs::op_node_fs_exists_sync<P>,
+    ops::fs::op_node_cp_sync<P>,
+    ops::fs::op_node_cp<P>,
     ops::winerror::op_node_sys_to_uv_error,
     ops::v8::op_v8_cached_data_version_tag,
     ops::v8::op_v8_get_heap_statistics,
@@ -329,6 +343,7 @@ deno_core::extension!(deno_node,
     "_fs/_fs_common.ts",
     "_fs/_fs_constants.ts",
     "_fs/_fs_copy.ts",
+    "_fs/_fs_cp.js",
     "_fs/_fs_dir.ts",
     "_fs/_fs_dirent.ts",
     "_fs/_fs_exists.ts",
diff --git a/ext/node/ops/fs.rs b/ext/node/ops/fs.rs
index 8e4805f6c..c5ae2371e 100644
--- a/ext/node/ops/fs.rs
+++ b/ext/node/ops/fs.rs
@@ -1,6 +1,9 @@
 // Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
 
+use std::cell::RefCell;
+use std::path::Path;
 use std::path::PathBuf;
+use std::rc::Rc;
 
 use deno_core::error::AnyError;
 use deno_core::op2;
@@ -24,3 +27,54 @@ where
   let fs = state.borrow::<FileSystemRc>();
   Ok(fs.lstat_sync(&path).is_ok())
 }
+
+#[op2(fast)]
+pub fn op_node_cp_sync<P>(
+  state: &mut OpState,
+  #[string] path: &str,
+  #[string] new_path: &str,
+) -> Result<(), AnyError>
+where
+  P: NodePermissions + 'static,
+{
+  let path = Path::new(path);
+  let new_path = Path::new(new_path);
+
+  state
+    .borrow_mut::<P>()
+    .check_read_with_api_name(path, Some("node:fs.cpSync"))?;
+  state
+    .borrow_mut::<P>()
+    .check_write_with_api_name(new_path, Some("node:fs.cpSync"))?;
+
+  let fs = state.borrow::<FileSystemRc>();
+  fs.cp_sync(path, new_path)?;
+  Ok(())
+}
+
+#[op2(async)]
+pub async fn op_node_cp<P>(
+  state: Rc<RefCell<OpState>>,
+  #[string] path: String,
+  #[string] new_path: String,
+) -> Result<(), AnyError>
+where
+  P: NodePermissions + 'static,
+{
+  let path = PathBuf::from(path);
+  let new_path = PathBuf::from(new_path);
+
+  let fs = {
+    let mut state = state.borrow_mut();
+    state
+      .borrow_mut::<P>()
+      .check_read_with_api_name(&path, Some("node:fs.cpSync"))?;
+    state
+      .borrow_mut::<P>()
+      .check_write_with_api_name(&new_path, Some("node:fs.cpSync"))?;
+    state.borrow::<FileSystemRc>().clone()
+  };
+
+  fs.cp_async(path, new_path).await?;
+  Ok(())
+}
diff --git a/ext/node/polyfills/_fs/_fs_cp.js b/ext/node/polyfills/_fs/_fs_cp.js
new file mode 100644
index 000000000..dbe327974
--- /dev/null
+++ b/ext/node/polyfills/_fs/_fs_cp.js
@@ -0,0 +1,41 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+
+// deno-lint-ignore-file prefer-primordials
+
+import {
+  getValidatedPath,
+  validateCpOptions,
+} from "ext:deno_node/internal/fs/utils.mjs";
+import { promisify } from "ext:deno_node/internal/util.mjs";
+
+const core = globalThis.__bootstrap.core;
+const ops = core.ops;
+const { op_node_cp } = core.ensureFastOps();
+
+export function cpSync(src, dest, options) {
+  validateCpOptions(options);
+  const srcPath = getValidatedPath(src, "src");
+  const destPath = getValidatedPath(dest, "dest");
+
+  ops.op_node_cp_sync(srcPath, destPath);
+}
+
+export function cp(src, dest, options, callback) {
+  if (typeof options === "function") {
+    callback = options;
+    options = {};
+  }
+  validateCpOptions(options);
+  const srcPath = getValidatedPath(src, "src");
+  const destPath = getValidatedPath(dest, "dest");
+
+  op_node_cp(
+    srcPath,
+    destPath,
+  ).then(
+    (res) => callback(null, res),
+    (err) => callback(err, null),
+  );
+}
+
+export const cpPromise = promisify(cp);
diff --git a/ext/node/polyfills/fs.ts b/ext/node/polyfills/fs.ts
index 881f0c139..01ac9912e 100644
--- a/ext/node/polyfills/fs.ts
+++ b/ext/node/polyfills/fs.ts
@@ -18,6 +18,7 @@ import {
   copyFilePromise,
   copyFileSync,
 } from "ext:deno_node/_fs/_fs_copy.ts";
+import { cp, cpPromise, cpSync } from "ext:deno_node/_fs/_fs_cp.js";
 import Dir from "ext:deno_node/_fs/_fs_dir.ts";
 import Dirent from "ext:deno_node/_fs/_fs_dirent.ts";
 import { exists, existsSync } from "ext:deno_node/_fs/_fs_exists.ts";
@@ -137,6 +138,7 @@ const {
 const promises = {
   access: accessPromise,
   copyFile: copyFilePromise,
+  cp: cpPromise,
   open: openPromise,
   opendir: opendirPromise,
   rename: renamePromise,
@@ -179,6 +181,8 @@ export default {
   constants,
   copyFile,
   copyFileSync,
+  cp,
+  cpSync,
   createReadStream,
   createWriteStream,
   Dir,
@@ -280,6 +284,8 @@ export {
   constants,
   copyFile,
   copyFileSync,
+  cp,
+  cpSync,
   createReadStream,
   createWriteStream,
   Dir,