[Turbopack] no need to depend on write completion (#69660)

### What

The node.js pool doesn't need to be invalidated when pool code is
re-writting. It's enough to invalidate it when the pool code has
changed.

Avoid having transient tasks in the embedded filesystem. It's not allow
to have persistent tasks depend on transient tasks with Persistent
Caching.

Avoid triggering invalidation when State is dropped. State might be
dropped after serialization when it's stored in persistent cache. This
doesn't mean we want to invalidate the tasks

crates/next-api/src/project.rs

@@ -1111,7 +1111,7 @@ impl Project {
     pub async fn emit_all_output_assets(
         self: Vc<Self>,
         output_assets: Vc<OutputAssetsOperation>,
-    ) -> Result<Vc<Completion>> {
+    ) -> Result<Vc<()>> {
         let span = tracing::info_span!("emitting");
         async move {
             let all_output_assets = all_assets_from_entries_operation(output_assets);
@@ -1120,21 +1120,27 @@ impl Project {
             let node_root = self.node_root();
 
             if let Some(map) = self.await?.versioned_content_map {
-                let completion = map.insert_output_assets(
-                    all_output_assets,
-                    node_root,
-                    client_relative_path,
-                    node_root,
-                );
-
-                Ok(completion)
+                let _ = map
+                    .insert_output_assets(
+                        all_output_assets,
+                        node_root,
+                        client_relative_path,
+                        node_root,
+                    )
+                    .resolve()
+                    .await?;
+
+                Ok(Vc::cell(()))
             } else {
-                Ok(emit_assets(
+                let _ = emit_assets(
                     *all_output_assets.await?,
                     node_root,
                     client_relative_path,
                     node_root,
-                ))
+                )
+                .resolve()
+                .await?;
+                Ok(Vc::cell(()))
             }
         }
         .instrument(span)

turbopack/crates/turbo-tasks-fs/src/embed/dir.rs

@@ -2,7 +2,7 @@ pub use ::include_dir::{
     include_dir, {self},
 };
 use anyhow::Result;
-use turbo_tasks::{RcStr, TransientInstance, Vc};
+use turbo_tasks::{RcStr, Vc};
 
 use crate::{embed::EmbeddedFileSystem, DiskFileSystem, FileSystem};
 
@@ -17,12 +17,11 @@ pub async fn directory_from_relative_path(
     Ok(Vc::upcast(disk_fs))
 }
 
-#[turbo_tasks::function]
-pub async fn directory_from_include_dir(
+pub fn directory_from_include_dir(
     name: RcStr,
-    dir: TransientInstance<&'static include_dir::Dir<'static>>,
-) -> Result<Vc<Box<dyn FileSystem>>> {
-    Ok(Vc::upcast(EmbeddedFileSystem::new(name, dir)))
+    dir: &'static include_dir::Dir<'static>,
+) -> Vc<Box<dyn FileSystem>> {
+    Vc::upcast(EmbeddedFileSystem::new(name, dir))
 }
 
 /// Returns an embedded [Vc<Box<dyn FileSystem>>] for the given path.
@@ -71,9 +70,6 @@ macro_rules! embed_directory_internal {
 
         static dir: include_dir::Dir<'static> = turbo_tasks_fs::embed::include_dir!($path);
 
-        turbo_tasks_fs::embed::directory_from_include_dir(
-            $name.into(),
-            turbo_tasks::TransientInstance::new(&dir),
-        )
+        turbo_tasks_fs::embed::directory_from_include_dir($name.into(), &dir)
     }};
 }

turbopack/crates/turbo-tasks-fs/src/embed/fs.rs

@@ -1,26 +1,21 @@
 use anyhow::{bail, Result};
 use include_dir::{Dir, DirEntry};
-use turbo_tasks::{Completion, RcStr, TransientInstance, ValueToString, Vc};
+use turbo_tasks::{Completion, RcStr, ValueToString, Vc};
 
 use crate::{
     DirectoryContent, DirectoryEntry, File, FileContent, FileMeta, FileSystem, FileSystemPath,
     LinkContent,
 };
 
-#[turbo_tasks::value(serialization = "none")]
+#[turbo_tasks::value(serialization = "none", cell = "new", eq = "manual")]
 pub struct EmbeddedFileSystem {
     name: RcStr,
     #[turbo_tasks(trace_ignore)]
-    dir: TransientInstance<&'static Dir<'static>>,
+    dir: &'static Dir<'static>,
 }
 
-#[turbo_tasks::value_impl]
 impl EmbeddedFileSystem {
-    #[turbo_tasks::function]
-    pub(super) fn new(
-        name: RcStr,
-        dir: TransientInstance<&'static Dir<'static>>,
-    ) -> Vc<EmbeddedFileSystem> {
+    pub(super) fn new(name: RcStr, dir: &'static Dir<'static>) -> Vc<EmbeddedFileSystem> {
         EmbeddedFileSystem { name, dir }.cell()
     }
 }
@@ -46,7 +41,7 @@ impl FileSystem for EmbeddedFileSystem {
     async fn read_dir(&self, path: Vc<FileSystemPath>) -> Result<Vc<DirectoryContent>> {
         let path_str = &path.await?.path;
         let dir = match (path_str.as_str(), self.dir.get_dir(path_str)) {
-            ("", _) => *self.dir,
+            ("", _) => self.dir,
             (_, Some(dir)) => dir,
             (_, None) => return Ok(DirectoryContent::NotFound.cell()),
         };

turbopack/crates/turbo-tasks-fs/src/invalidator_map.rs

@@ -71,16 +71,3 @@ impl<'de> Deserialize<'de> for InvalidatorMap {
         deserializer.deserialize_newtype_struct("InvalidatorMap", V)
     }
 }
-
-impl Drop for InvalidatorMap {
-    fn drop(&mut self) {
-        while let Ok((_, value)) = self.queue.pop() {
-            value.invalidate();
-        }
-        for (_, invalidators) in self.map.lock().unwrap().drain() {
-            for invalidator in invalidators {
-                invalidator.invalidate();
-            }
-        }
-    }
-}

turbopack/crates/turbo-tasks/src/state.rs

@@ -68,15 +68,6 @@ impl<'de, T> Deserialize<'de> for State<T> {
     }
 }
 
-impl<T> Drop for State<T> {
-    fn drop(&mut self) {
-        let mut inner = self.inner.lock();
-        for invalidator in take(&mut inner.invalidators) {
-            invalidator.invalidate();
-        }
-    }
-}
-
 impl<T> State<T> {
     pub fn new(value: T) -> Self {
         mark_stateful();

turbopack/crates/turbopack-node/src/evaluate.rs

@@ -21,6 +21,7 @@ use turbo_tasks_env::ProcessEnv;
 use turbo_tasks_fs::{to_sys_path, File, FileSystemPath};
 use turbopack_core::{
     asset::AssetContent,
+    changed::content_changed,
     chunk::{ChunkingContext, ChunkingContextExt, EvaluatableAsset, EvaluatableAssets},
     context::AssetContext,
     error::PrettyPrintError,
@@ -153,11 +154,11 @@ pub async fn get_evaluate_pool(
         chunking_context.root_entry_chunk_group_asset(path, entry_module, runtime_entries);
 
     let output_root: Vc<FileSystemPath> = chunking_context.output_root();
-    let emit_package = emit_package_json(output_root);
-    let emit = emit(bootstrap, output_root);
+    let _ = emit_package_json(output_root);
+    // Invalidate pool when code content changes
+    content_changed(Vc::upcast(bootstrap)).await?;
+    let _ = emit(bootstrap, output_root);
     let assets_for_source_mapping = internal_assets_for_source_mapping(bootstrap, output_root);
-    emit_package.await?;
-    emit.await?;
     let pool = NodeJsPool::new(
         cwd,
         entrypoint,