Repository: vercel/next.js
Pull request: Turbopack: Define `Effect` as a trait instead of a closure

This switches effects (special `Collectible`s used to defer file writes) from using a closure representation to using a trait representation.

This makes constructing effects a little more annoying, but it ensures that we'll be able to correctly implement `TraceRawVcs` on these in a subsequent PR.

Changes to `turbo-tasks/src/effect.rs` were done by hand, an LLM was used for updating the two callsites.

Changed files:
filename: turbopack/crates/turbo-tasks-fs/src/lib.rs
patch: @@ -36,6 +36,7 @@ use std::{
     borrow::Cow,
     cmp::{Ordering, min},
     env,
+    error::Error as StdError,
     fmt::{self, Debug, Formatter},
     fs::FileType,
     future::Future,
@@ -63,9 +64,9 @@ use tokio::{
 use tracing::Instrument;
 use turbo_rcstr::{RcStr, rcstr};
 use turbo_tasks::{
-    ApplyEffectsContext, Completion, InvalidationReason, Invalidator, NonLocalValue, ReadRef,
-    ResolvedVc, TaskInput, TurboTasksApi, ValueToString, ValueToStringRef, Vc,
-    debug::ValueDebugFormat, effect, mark_session_dependent, parallel, trace::TraceRawVcs,
+    ApplyEffectsContext, Completion, Effect, InvalidationReason, Invalidator, NonLocalValue,
+    ReadRef, ResolvedVc, TaskInput, TurboTasksApi, ValueToString, ValueToStringRef, Vc,
+    debug::ValueDebugFormat, emit_effect, mark_session_dependent, parallel, trace::TraceRawVcs,
     turbo_tasks_weak, turbobail, turbofmt,
 };
 use turbo_tasks_hash::{
@@ -975,114 +976,136 @@ impl FileSystem for DiskFileSystem {
         let inner = self.inner.clone();
         let invalidator = turbo_tasks::get_invalidator();
 
-        effect(async move {
-            let full_path = validate_path_length(&full_path)?;
+        struct WriteEffect {
+            full_path: PathBuf,
+            inner: Arc<DiskFileSystemInner>,
+            invalidator: Option<Invalidator>,
+            content: ReadRef<FileContent>,
+        }
 
-            let _lock = inner.lock_path(&full_path).await;
+        impl Effect for WriteEffect {
+            type Error = AnyhowWrapper;
 
-            // Track the file, so that we will rewrite it if it ever changes.
-            let old_invalidators = invalidator
-                .map(|invalidator| {
-                    inner.register_write_invalidator(
-                        &full_path,
-                        invalidator,
-                        WriteContent::File(content.clone()),
-                    )
-                })
-                .transpose()?
-                .unwrap_or_default();
-
-            // We perform an untracked comparison here, so that this write is not dependent
-            // on a read's Vc<FileContent> (and the memory it holds). Our untracked read can
-            // be freed immediately. Given this is an output file, it's unlikely any Turbo
-            // code will need to read the file from disk into a Vc<FileContent>, so we're
-            // not wasting cycles.
-            let compare = content
-                .streaming_compare(&full_path)
-                .instrument(tracing::info_span!("read file before write", name = ?full_path))
-                .concurrency_limited(&inner.read_semaphore)
-                .await?;
-            if compare == FileComparison::Equal {
-                if !old_invalidators.is_empty() {
-                    for (invalidator, write_content) in old_invalidators {
-                        inner.invalidator_map.insert(
-                            full_path.clone().into_owned(),
+            async fn apply(self) -> Result<(), Self::Error> {
+                let full_path = validate_path_length(&self.full_path)?;
+
+                let _lock = self.inner.lock_path(&full_path).await;
+
+                // Track the file, so that we will rewrite it if it ever changes.
+                let old_invalidators = self
+                    .invalidator
+                    .map(|invalidator| {
+                        self.inner.register_write_invalidator(
+                            &full_path,
                             invalidator,
-                            write_content,
-                        );
+                            WriteContent::File(self.content.clone()),
+                        )
+                    })
+                    .transpose()?
+                    .unwrap_or_default();
+
+                // We perform an untracked comparison here, so that this write is not dependent
+                // on a read's Vc<FileContent> (and the memory it holds). Our untracked read can
+                // be freed immediately. Given this is an output file, it's unlikely any Turbo
+                // code will need to read the file from disk into a Vc<FileContent>, so we're
+                // not wasting cycles.
+                let compare = self
+                    .content
+                    .streaming_compare(&full_path)
+                    .instrument(tracing::info_span!("read file before write", name = ?full_path))
+                    .concurrency_limited(&self.inner.read_semaphore)
+                    .await?;
+                if compare == FileComparison::Equal {
+                    if !old_invalidators.is_empty() {
+                        for (invalidator, write_content) in old_invalidators {
+                            self.inner.invalidator_map.insert(
+                                full_path.clone().into_owned(),
+                                invalidator,
+                                write_content,
+                            );
+                        }
                     }
+                    return Ok(());
                 }
-                return Ok(());
-            }
 
-            match &*content {
-                FileContent::Content(..) => {
-                    let create_directory = compare == FileComparison::Create;
-                    if create_directory && let Some(parent) = full_path.parent() {
-                        inner.create_directory(parent).await.with_context(|| {
-                            format!(
-                                "failed to create directory {parent:?} for write to {full_path:?}",
-                            )
-                        })?;
-                    }
+                match &*self.content {
+                    FileContent::Content(..) => {
+                        let create_directory = compare == FileComparison::Create;
+                        if create_directory && let Some(parent) = full_path.parent() {
+                            self.inner.create_directory(parent).await.with_context(|| {
+                                format!(
+                                    "failed to create directory {parent:?} for write to \
+                                     {full_path:?}",
+                                )
+                            })?;
+                        }
 
-                    let content = content.clone();
-                    retry_blocking(|| {
-                        let mut f = std::fs::File::create(&full_path)?;
-                        let FileContent::Content(file) = &*content else {
-                            unreachable!()
-                        };
-                        std::io::copy(&mut file.read(), &mut f)?;
-                        #[cfg(unix)]
-                        f.set_permissions(file.meta.permissions.into())?;
-                        f.flush()?;
-
-                        static WRITE_VERSION: LazyLock<bool> = LazyLock::new(|| {
-                            std::env::var_os("TURBO_ENGINE_WRITE_VERSION")
-                                .is_some_and(|v| v == "1" || v == "true")
-                        });
-                        if *WRITE_VERSION {
-                            let mut full_path = full_path.clone().into_owned();
-                            let hash = hash_xxh3_hash64(file);
-                            let ext = full_path.extension();
-                            let ext = if let Some(ext) = ext {
-                                format!("{:016x}.{}", hash, ext.to_string_lossy())
-                            } else {
-                                format!("{hash:016x}")
-                            };
-                            full_path.set_extension(ext);
+                        let content = self.content.clone();
+                        retry_blocking(|| {
                             let mut f = std::fs::File::create(&full_path)?;
+                            let FileContent::Content(file) = &*content else {
+                                unreachable!()
+                            };
                             std::io::copy(&mut file.read(), &mut f)?;
                             #[cfg(unix)]
                             f.set_permissions(file.meta.permissions.into())?;
                             f.flush()?;
-                        }
-                        Ok::<(), io::Error>(())
-                    })
-                    .instrument(tracing::info_span!("write file", name = ?full_path))
-                    .concurrency_limited(&inner.write_semaphore)
-                    .await
-                    .with_context(|| format!("failed to write to {full_path:?}"))?;
-                }
-                FileContent::NotFound => {
-                    retry_blocking(|| std::fs::remove_file(&full_path))
-                        .instrument(tracing::info_span!("remove file", name = ?full_path))
-                        .concurrency_limited(&inner.write_semaphore)
-                        .await
-                        .or_else(|err| {
-                            if err.kind() == ErrorKind::NotFound {
-                                Ok(())
-                            } else {
-                                Err(err)
+
+                            static WRITE_VERSION: LazyLock<bool> = LazyLock::new(|| {
+                                std::env::var_os("TURBO_ENGINE_WRITE_VERSION")
+                                    .is_some_and(|v| v == "1" || v == "true")
+                            });
+                            if *WRITE_VERSION {
+                                let mut full_path = full_path.clone().into_owned();
+                                let hash = hash_xxh3_hash64(file);
+                                let ext = full_path.extension();
+                                let ext = if let Some(ext) = ext {
+                                    format!("{:016x}.{}", hash, ext.to_string_lossy())
+                                } else {
+                                    format!("{hash:016x}")
+                                };
+                                full_path.set_extension(ext);
+                                let mut f = std::fs::File::create(&full_path)?;
+                                std::io::copy(&mut file.read(), &mut f)?;
+                                #[cfg(unix)]
+                                f.set_permissions(file.meta.permissions.into())?;
+                                f.flush()?;
                             }
+                            Ok::<(), io::Error>(())
                         })
-                        .with_context(|| format!("removing {full_path:?} failed"))?;
+                        .instrument(tracing::info_span!("write file", name = ?full_path))
+                        .concurrency_limited(&self.inner.write_semaphore)
+                        .await
+                        .with_context(|| format!("failed to write to {full_path:?}"))?;
+                    }
+                    FileContent::NotFound => {
+                        retry_blocking(|| std::fs::remove_file(&full_path))
+                            .instrument(tracing::info_span!("remove file", name = ?full_path))
+                            .concurrency_limited(&self.inner.write_semaphore)
+                            .await
+                            .or_else(|err| {
+                                if err.kind() == ErrorKind::NotFound {
+                                    Ok(())
+                                } else {
+                                    Err(err)
+                                }
+                            })
+                            .with_context(|| format!("removing {full_path:?} failed"))?;
+                    }
                 }
-            }
 
-            inner.invalidate_from_write(&full_path, old_invalidators);
+                self.inner
+                    .invalidate_from_write(&full_path, old_invalidators);
 
-            Ok(())
+                Ok(())
+            }
+        }
+
+        emit_effect(WriteEffect {
+            full_path,
+            inner,
+            invalidator,
+            content,
         });
 
         Ok(())
@@ -1105,196 +1128,219 @@ impl FileSystem for DiskFileSystem {
         let inner = self.inner.clone();
         let invalidator = turbo_tasks::get_invalidator();
 
-        effect(async move {
-            let full_path = validate_path_length(&full_path)?;
+        struct WriteLinkEffect {
+            full_path: PathBuf,
+            inner: Arc<DiskFileSystemInner>,
+            invalidator: Option<Invalidator>,
+            content: ReadRef<LinkContent>,
+        }
+
+        impl Effect for WriteLinkEffect {
+            type Error = AnyhowWrapper;
 
-            let _lock = inner.lock_path(&full_path).await;
+            async fn apply(self) -> Result<(), Self::Error> {
+                let full_path = validate_path_length(&self.full_path)?;
 
-            let old_invalidators = invalidator
-                .map(|invalidator| {
-                    inner.register_write_invalidator(
-                        &full_path,
-                        invalidator,
-                        WriteContent::Link(content.clone()),
-                    )
-                })
-                .transpose()?
-                .unwrap_or_default();
-
-            enum OsSpecificLinkContent {
-                Link {
-                    #[cfg(windows)]
-                    is_directory: bool,
-                    target: PathBuf,
-                },
-                NotFound,
-                Invalid,
-            }
+                let _lock = self.inner.lock_path(&full_path).await;
 
-            let os_specific_link_content = match &*content {
-                LinkContent::Link { target, link_type } => {
-                    let is_directory = link_type.contains(LinkType::DIRECTORY);
-                    let target_path = if link_type.contains(LinkType::ABSOLUTE) {
-                        Path::new(&inner.root).join(unix_to_sys(target).as_ref())
-                    } else {
-                        let relative_target = PathBuf::from(unix_to_sys(target).as_ref());
-                        if cfg!(windows) && is_directory {
-                            // Windows junction points must always be stored as absolute
-                            full_path
-                                .parent()
-                                .unwrap_or(&full_path)
-                                .join(relative_target)
+                let old_invalidators = self
+                    .invalidator
+                    .map(|invalidator| {
+                        self.inner.register_write_invalidator(
+                            &full_path,
+                            invalidator,
+                            WriteContent::Link(self.content.clone()),
+                        )
+                    })
+                    .transpose()?
+                    .unwrap_or_default();
+
+                enum OsSpecificLinkContent {
+                    Link {
+                        #[cfg(windows)]
+                        is_directory: bool,
+                        target: PathBuf,
+                    },
+                    NotFound,
+                    Invalid,
+                }
+
+                let os_specific_link_content = match &*self.content {
+                    LinkContent::Link { target, link_type } => {
+                        let is_directory = link_type.contains(LinkType::DIRECTORY);
+                        let target_path = if link_type.contains(LinkType::ABSOLUTE) {
+                            Path::new(&self.inner.root).join(unix_to_sys(target).as_ref())
                         } else {
-                            relative_target
+                            let relative_target = PathBuf::from(unix_to_sys(target).as_ref());
+                            if cfg!(windows) && is_directory {
+                                // Windows junction points must always be stored as absolute
+                                full_path
+                                    .parent()
+                                    .unwrap_or(&full_path)
+                                    .join(relative_target)
+                            } else {
+                                relative_target
+                            }
+                        };
+                        OsSpecificLinkContent::Link {
+                            #[cfg(windows)]
+                            is_directory,
+                            target: target_path,
                         }
-                    };
-                    OsSpecificLinkContent::Link {
-                        #[cfg(windows)]
-                        is_directory,
-                        target: target_path,
                     }
-                }
-                LinkContent::Invalid => OsSpecificLinkContent::Invalid,
-                LinkContent::NotFound => OsSpecificLinkContent::NotFound,
-            };
+                    LinkContent::Invalid => OsSpecificLinkContent::Invalid,
+                    LinkContent::NotFound => OsSpecificLinkContent::NotFound,
+                };
 
-            let old_content = match retry_blocking(|| std::fs::read_link(&full_path))
-                .instrument(tracing::info_span!("read symlink before write", name = ?full_path))
-                .concurrency_limited(&inner.read_semaphore)
-                .await
-            {
-                Ok(res) => Some((res.is_absolute(), res)),
-                Err(_) => None,
-            };
-            let is_equal = match (&os_specific_link_content, &old_content) {
-                (
-                    OsSpecificLinkContent::Link { target, .. },
-                    Some((old_is_absolute, old_target)),
-                ) => target == old_target && target.is_absolute() == *old_is_absolute,
-                (OsSpecificLinkContent::NotFound, None) => true,
-                _ => false,
-            };
-            if is_equal {
-                if !old_invalidators.is_empty() {
-                    for (invalidator, write_content) in old_invalidators {
-                        inner.invalidator_map.insert(
-                            full_path.clone().into_owned(),
-                            invalidator,
-                            write_content,
-                        );
+                let old_content = match retry_blocking(|| std::fs::read_link(&full_path))
+                    .instrument(tracing::info_span!("read symlink before write", name = ?full_path))
+                    .concurrency_limited(&self.inner.read_semaphore)
+                    .await
+                {
+                    Ok(res) => Some((res.is_absolute(), res)),
+                    Err(_) => None,
+                };
+                let is_equal = match (&os_specific_link_content, &old_content) {
+                    (
+                        OsSpecificLinkContent::Link { target, .. },
+                        Some((old_is_absolute, old_target)),
+                    ) => target == old_target && target.is_absolute() == *old_is_absolute,
+                    (OsSpecificLinkContent::NotFound, None) => true,
+                    _ => false,
+                };
+                if is_equal {
+                    if !old_invalidators.is_empty() {
+                        for (invalidator, write_content) in old_invalidators {
+                            self.inner.invalidator_map.insert(
+                                full_path.clone().into_owned(),
+                                invalidator,
+                                write_content,
+                            );
+                        }
                     }
+                    return Ok(());
                 }
-                return Ok(());
-            }
 
-            match os_specific_link_content {
-                OsSpecificLinkContent::Link {
-                    target,
-                    #[cfg(windows)]
-                    is_directory,
-                    ..
-                } => {
-                    let full_path = full_path.into_owned();
-
-                    let create_directory = old_content.is_none();
-                    if create_directory && let Some(parent) = full_path.parent() {
-                        inner.create_directory(parent).await.with_context(|| {
-                            format!(
-                                "failed to create directory {parent:?} for write link to \
-                                 {full_path:?}",
-                            )
-                        })?;
-                    }
+                match os_specific_link_content {
+                    OsSpecificLinkContent::Link {
+                        target,
+                        #[cfg(windows)]
+                        is_directory,
+                        ..
+                    } => {
+                        let full_path = full_path.into_owned();
+
+                        let create_directory = old_content.is_none();
+                        if create_directory && let Some(parent) = full_path.parent() {
+                            self.inner.create_directory(parent).await.with_context(|| {
+                                format!(
+                                    "failed to create directory {parent:?} for write link to \
+                                     {full_path:?}",
+                                )
+                            })?;
+                        }
 
-                    #[derive(thiserror::Error, Debug)]
-                    #[error("{msg}: {source}")]
-                    struct SymlinkCreationError {
-                        msg: &'static str,
-                        #[source]
-                        source: io::Error,
-                    }
+                        #[derive(thiserror::Error, Debug)]
+                        #[error("{msg}: {source}")]
+                        struct SymlinkCreationError {
+                            msg: &'static str,
+                            #[source]
+                            source: io::Error,
+                        }
 
-                    let mut has_old_content = old_content.is_some();
-                    let try_create_link = || {
-                        if has_old_content {
-                            // Remove existing symlink before creating a new one. On Unix,
-                            // symlink(2) fails with EEXIST if the link already exists instead of
-                            // overwriting it. Windows has similar behavior with junction points.
-                            remove_symbolic_link_dir_helper(&full_path).map_err(|err| {
+                        let mut has_old_content = old_content.is_some();
+                        let try_create_link = || {
+                            if has_old_content {
+                                // Remove existing symlink before creating a new one. On Unix,
+                                // symlink(2) fails with EEXIST if the link already exists instead
+                                // of overwriting it. Windows has similar behavior with junction
+                                // points.
+                                remove_symbolic_link_dir_helper(&full_path).map_err(|err| {
+                                    SymlinkCreationError {
+                                        msg: "removal of existing symbolic link or junction point \
+                                              failed",
+                                        source: err,
+                                    }
+                                })?;
+                                has_old_content = false;
+                            }
+                            #[cfg(not(windows))]
+                            let io_result = std::os::unix::fs::symlink(&target, &full_path);
+                            #[cfg(windows)]
+                            let io_result = if is_directory {
+                                std::os::windows::fs::junction_point(&target, &full_path)
+                            } else {
+                                std::os::windows::fs::symlink_file(&target, &full_path)
+                            };
+                            io_result.map_err(|err| {
+                                if err.kind() == ErrorKind::AlreadyExists {
+                                    // try to remove the symlink on the next iteration of the loop
+                                    has_old_content = true;
+                                }
                                 SymlinkCreationError {
-                                    msg: "removal of existing symbolic link or junction point \
-                                          failed",
+                                    msg: "creation of a new symbolic link or junction point failed",
                                     source: err,
                                 }
-                            })?;
-                            has_old_content = false;
+                            })
+                        };
+                        fn can_retry_link(err: &SymlinkCreationError) -> bool {
+                            err.source.kind() == ErrorKind::AlreadyExists || can_retry(&err.source)
                         }
-                        #[cfg(not(windows))]
-                        let io_result = std::os::unix::fs::symlink(&target, &full_path);
-                        #[cfg(windows)]
-                        let io_result = if is_directory {
-                            std::os::windows::fs::junction_point(&target, &full_path)
-                        } else {
-                            std::os::windows::fs::symlink_file(&target, &full_path)
+                        let err_context = || {
+                            #[cfg(not(windows))]
+                            let message = format!(
+                                "failed to create symlink at {full_path:?} pointing to {target:?}"
+                            );
+                            #[cfg(windows)]
+                            let message = if is_directory {
+                                format!(
+                                    "failed to create junction point at {full_path:?} pointing to \
+                                     {target:?}"
+                                )
+                            } else {
+                                format!(
+                                    "failed to create symlink at {full_path:?} pointing to \
+                                     {target:?}\n\
+                                    (Note: creating file symlinks on Windows require developer \
+                                     mode or admin permissions: \
+                                     https://learn.microsoft.com/en-us/windows/advanced-settings/developer-mode)",
+                                )
+                            };
+                            message
                         };
-                        io_result.map_err(|err| {
-                            if err.kind() == ErrorKind::AlreadyExists {
-                                // try to remove the symlink on the next iteration of the loop
-                                has_old_content = true;
-                            }
-                            SymlinkCreationError {
-                                msg: "creation of a new symbolic link or junction point failed",
-                                source: err,
-                            }
-                        })
-                    };
-                    fn can_retry_link(err: &SymlinkCreationError) -> bool {
-                        err.source.kind() == ErrorKind::AlreadyExists || can_retry(&err.source)
+                        retry_blocking_custom(try_create_link, can_retry_link)
+                            .instrument(tracing::info_span!(
+                                "write symlink",
+                                name = ?full_path,
+                                target = ?target,
+                            ))
+                            .concurrency_limited(&self.inner.write_semaphore)
+                            .await
+                            .with_context(err_context)?;
+                    }
+                    OsSpecificLinkContent::Invalid => {
+                        return Err(AnyhowWrapper(anyhow!(
+                            "invalid symlink target: {full_path:?}"
+                        )));
+                    }
+                    OsSpecificLinkContent::NotFound => {
+                        retry_blocking(|| remove_symbolic_link_dir_helper(&full_path))
+                            .instrument(tracing::info_span!("remove symlink", name = ?full_path))
+                            .concurrency_limited(&self.inner.write_semaphore)
+                            .await
+                            .with_context(|| format!("removing {full_path:?} failed"))?;
                     }
-                    let err_context = || {
-                        #[cfg(not(windows))]
-                        let message = format!(
-                            "failed to create symlink at {full_path:?} pointing to {target:?}"
-                        );
-                        #[cfg(windows)]
-                        let message = if is_directory {
-                            format!(
-                                "failed to create junction point at {full_path:?} pointing to \
-                                 {target:?}"
-                            )
-                        } else {
-                            format!(
-                                "failed to create symlink at {full_path:?} pointing to {target:?}\n\
-                                (Note: creating file symlinks on Windows require developer mode or \
-                                admin permissions: \
-                                https://learn.microsoft.com/en-us/windows/advanced-settings/developer-mode)",
-                            )
-                        };
-                        message
-                    };
-                    retry_blocking_custom(try_create_link, can_retry_link)
-                        .instrument(tracing::info_span!(
-                            "write symlink",
-                            name = ?full_path,
-                            target = ?target,
-                        ))
-                        .concurrency_limited(&inner.write_semaphore)
-                        .await
-                        .with_context(err_context)?;
-                }
-                OsSpecificLinkContent::Invalid => {
-                    bail!("invalid symlink target: {full_path:?}")
-                }
-                OsSpecificLinkContent::NotFound => {
-                    retry_blocking(|| remove_symbolic_link_dir_helper(&full_path))
-                        .instrument(tracing::info_span!("remove symlink", name = ?full_path))
-                        .concurrency_limited(&inner.write_semaphore)
-                        .await
-                        .with_context(|| format!("removing {full_path:?} failed"))?;
                 }
+
+                Ok(())
             }
+        }
 
-            Ok(())
+        emit_effect(WriteLinkEffect {
+            full_path,
+            inner,
+            invalidator,
+            content,
         });
         Ok(())
     }
@@ -2815,6 +2861,34 @@ async fn realpath_with_links(path: FileSystemPath) -> Result<Vc<RealPathResult>>
     .cell())
 }
 
+/// Wrapper to convert `anyhow::Error` to `impl std::error::Error` for use in `Effect::apply`.
+// TODO(bgw): use a structured error type instead of anyhow for write/write_link
+struct AnyhowWrapper(anyhow::Error);
+
+impl std::fmt::Display for AnyhowWrapper {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        std::fmt::Display::fmt(&self.0, f)
+    }
+}
+
+impl std::fmt::Debug for AnyhowWrapper {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        std::fmt::Debug::fmt(&self.0, f)
+    }
+}
+
+impl StdError for AnyhowWrapper {
+    fn source(&self) -> Option<&(dyn StdError + 'static)> {
+        self.0.source()
+    }
+}
+
+impl From<anyhow::Error> for AnyhowWrapper {
+    fn from(err: anyhow::Error) -> Self {
+        AnyhowWrapper(err)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use turbo_rcstr::rcstr;

filename: turbopack/crates/turbo-tasks/src/effect.rs
patch: @@ -1,5 +1,6 @@
 use std::{
     any::{Any, TypeId},
+    error::Error as StdError,
     future::Future,
     mem::replace,
     panic,
@@ -19,31 +20,66 @@ use crate::{
     self as turbo_tasks, CollectiblesSource, ReadRef, ResolvedVc, TryJoinIterExt, emit,
     event::{Event, EventListener},
     spawn,
-    util::SharedError,
 };
 
 const APPLY_EFFECTS_CONCURRENCY_LIMIT: usize = 1024;
 
-/// A trait to emit a task effect as collectible. This trait only has one
-/// implementation, `EffectInstance` and no other implementation is allowed.
-/// The trait is private to this module so that no other implementation can be
-/// added.
-#[turbo_tasks::value_trait]
-trait Effect {}
+pub trait Effect: Send + Sync + 'static {
+    type Error: StdError + Send + Sync + 'static;
 
-/// A future that represents the effect of a task. The future is executed when
-/// the effect is applied.
-type EffectFuture = Pin<Box<dyn Future<Output = Result<()>> + Send + Sync + 'static>>;
+    /// A function that is called once at the top level of the program's execution after everything
+    /// has "settled".
+    ///
+    /// This function is executed outside of the turbo-tasks context, and therefore cannot read any
+    /// `Vc`s or call any turbo-task functions. The effect can store [`ResolvedVc`]s (or any other
+    /// `Vc` type), but should not read or resolve their contents.
+    fn apply(self) -> impl Future<Output = Result<(), Self::Error>> + Send;
+}
 
-/// The inner state of an effect instance if it has not been applied yet.
-struct EffectInner {
-    future: EffectFuture,
+/// The error type that an effect can return. We use `dyn std::error::Error` (instead of
+/// [`anyhow::Error`] or [`SharedError`]) to encourage use of structured error types that can
+/// potentially be transformed into `Issue`s.
+///
+/// We can't require that the returned error implements `Issue`:
+/// - `Issue` uses `FileSystemPath`
+/// - `turbo-tasks-fs` returns effect errors that should be transformed into `Issue`s.
+/// - It logically doesn't make sense to define `Issue` in `turbo-tasks-fs`, `Issue` can't be
+///   defined in a base crate either because it would form a circular crate dependency.
+///
+/// So instead, we leave it up to the caller to figure out how to downcast these errors themselves.
+///
+/// [`SharedError`]: crate::util::SharedError
+type EffectError = Arc<dyn StdError + Send + Sync + 'static>;
+
+// Private wrapper trait to allow dynamic dispatch of an `Effect`. This is similar to the pattern
+// that the dynosaur crate uses: https://github.com/spastorino/dynosaur
+trait DynEffect: Send + Sync + 'static {
+    fn dyn_apply(self: Box<Self>) -> Pin<Box<dyn Future<Output = Result<(), EffectError>> + Send>>;
 }
 
+impl<T> DynEffect for T
+where
+    T: Effect,
+{
+    fn dyn_apply(self: Box<Self>) -> Pin<Box<dyn Future<Output = Result<(), EffectError>> + Send>> {
+        Box::pin(async move {
+            self.apply()
+                .await
+                .map_err(|err| Arc::new(err) as EffectError)
+        })
+    }
+}
+
+/// A trait to emit a task effect as collectible. This trait only has one implementation,
+/// `EffectInstance` and no other implementation is allowed. The trait is private to this module so
+/// that no other implementation can be added.
+#[turbo_tasks::value_trait]
+trait EffectCollectible {}
+
 enum EffectState {
-    NotStarted(EffectInner),
+    NotStarted(Box<dyn DynEffect>),
     Started(Event),
-    Finished(Result<(), SharedError>),
+    Finished(Result<(), EffectError>),
 }
 
 /// The Effect instance collectible that is emitted for effects.
@@ -54,19 +90,19 @@ struct EffectInstance {
 }
 
 impl EffectInstance {
-    fn new(future: impl Future<Output = Result<()>> + Send + Sync + 'static) -> Self {
+    fn new(effect: impl Effect) -> Self {
         Self {
-            inner: Mutex::new(EffectState::NotStarted(EffectInner {
-                future: Box::pin(future),
-            })),
+            inner: Mutex::new(EffectState::NotStarted(
+                Box::new(effect) as Box<dyn DynEffect>
+            )),
         }
     }
 
     async fn apply(&self) -> Result<()> {
         loop {
             enum State {
                 Started(EventListener),
-                NotStarted(EffectInner),
+                NotStarted(Box<dyn DynEffect>),
             }
             let state = {
                 let mut guard = self.inner.lock();
@@ -93,10 +129,11 @@ impl EffectInstance {
                 State::Started(listener) => {
                     listener.await;
                 }
-                State::NotStarted(EffectInner { future }) => {
-                    let join_handle = spawn(ApplyEffectsContext::in_current_scope(future));
+                State::NotStarted(effect) => {
+                    let join_handle =
+                        spawn(ApplyEffectsContext::in_current_scope(effect.dyn_apply()));
                     let result = match join_handle.await {
-                        Err(err) => Err(SharedError::new(err)),
+                        Err(err) => Err(err),
                         Ok(()) => Ok(()),
                     };
                     let event = {
@@ -117,19 +154,19 @@ impl EffectInstance {
 }
 
 #[turbo_tasks::value_impl]
-impl Effect for EffectInstance {}
+impl EffectCollectible for EffectInstance {}
 
-/// Schedules an effect to be applied. The passed future is executed once `apply_effects` is called.
+/// Emits an effect to be applied. The effect is executed once `apply_effects` is called.
 ///
-/// The effect will only executed once. The passed future is executed outside of the current task
-/// and can't read any Vcs. These need to be read before. ReadRefs can be passed into the future.
+/// The effect will only executed once. The effect is executed outside of the current task
+/// and can't read any Vcs. These need to be read before. ReadRefs can be passed into the effect.
 ///
 /// Effects are executed in parallel, so they might need to use async locking to avoid problems.
 /// Order of execution of multiple effects is not defined. You must not use multiple conflicting
 /// effects to avoid non-deterministic behavior.
-pub fn effect(future: impl Future<Output = Result<()>> + Send + Sync + 'static) {
-    emit::<Box<dyn Effect>>(ResolvedVc::upcast(
-        EffectInstance::new(future).resolved_cell(),
+pub fn emit_effect(effect: impl Effect) {
+    emit::<Box<dyn EffectCollectible>>(ResolvedVc::upcast(
+        EffectInstance::new(effect).resolved_cell(),
     ));
 }
 
@@ -151,7 +188,7 @@ pub fn effect(future: impl Future<Output = Result<()>> + Send + Sync + 'static)
 /// apply_effects(operation).await?;
 /// ```
 pub async fn apply_effects(source: impl CollectiblesSource) -> Result<()> {
-    let effects: AutoSet<ResolvedVc<Box<dyn Effect>>> = source.take_collectibles();
+    let effects: AutoSet<ResolvedVc<Box<dyn EffectCollectible>>> = source.take_collectibles();
     if effects.is_empty() {
         return Ok(());
     }
@@ -194,7 +231,7 @@ pub async fn apply_effects(source: impl CollectiblesSource) -> Result<()> {
 /// result_with_effects.effects.apply().await?;
 /// ```
 pub async fn get_effects(source: impl CollectiblesSource) -> Result<Effects> {
-    let effects: AutoSet<ResolvedVc<Box<dyn Effect>>> = source.take_collectibles();
+    let effects: AutoSet<ResolvedVc<Box<dyn EffectCollectible>>> = source.take_collectibles();
     let effects = effects
         .into_iter()
         .map(|effect| async move {

filename: turbopack/crates/turbo-tasks/src/lib.rs
patch: @@ -78,7 +78,7 @@ pub use crate::{
     collectibles::CollectiblesSource,
     completion::{Completion, Completions},
     display::{ValueToString, ValueToStringRef},
-    effect::{ApplyEffectsContext, Effects, apply_effects, effect, get_effects},
+    effect::{ApplyEffectsContext, Effect, Effects, apply_effects, emit_effect, get_effects},
     error::PrettyPrintError,
     id::{ExecutionId, LocalTaskId, TRANSIENT_TASK_BIT, TaskId, TraitTypeId, ValueTypeId},
     invalidation::{