add helper for walking a tree concurrently and deterministic (vercel/turborepo#3619)

Similar to our `try_join` helper this adds a
`try_flat_map_recursive_join` helper, which allows to async expand a
tree structure into all nodes. It will call the async mapper function
concurrently to allow parallelism. It will handle circular and duplicate
references and return all nodes in a determinstic way (breath-first).

Author: sokra

crates/next-core/src/manifest.rs

@@ -1,7 +1,6 @@
 use anyhow::Result;
-use indexmap::IndexSet;
 use mime::APPLICATION_JSON;
-use turbo_tasks::primitives::StringsVc;
+use turbo_tasks::{primitives::StringsVc, TryFlatMapRecursiveJoinIterExt, TryJoinIterExt};
 use turbo_tasks_fs::File;
 use turbopack_core::asset::AssetContentVc;
 use turbopack_dev_server::source::{
@@ -24,31 +23,47 @@ impl DevManifestContentSourceVc {
     #[turbo_tasks::function]
     async fn find_routes(self) -> Result<StringsVc> {
         let this = &*self.await?;
-        let mut queue = this.page_roots.clone();
-        let mut routes = IndexSet::new();
-
-        while let Some(content_source) = queue.pop() {
-            queue.extend(content_source.get_children().await?.iter());
 
+        async fn content_source_to_pathname(
+            content_source: ContentSourceVc,
+        ) -> Result<Option<String>> {
             // TODO This shouldn't use casts but an public api instead
             if let Some(api_source) = NodeApiContentSourceVc::resolve_from(content_source).await? {
-                routes.insert(format!("/{}", api_source.get_pathname().await?));
-
-                continue;
+                return Ok(Some(format!("/{}", api_source.get_pathname().await?)));
             }
 
             if let Some(page_source) =
                 NodeRenderContentSourceVc::resolve_from(content_source).await?
             {
-                routes.insert(format!("/{}", page_source.get_pathname().await?));
-
-                continue;
+                return Ok(Some(format!("/{}", page_source.get_pathname().await?)));
             }
+
+            Ok(None)
         }
 
+        async fn get_content_source_children(
+            content_source: ContentSourceVc,
+        ) -> Result<Vec<ContentSourceVc>> {
+            Ok(content_source.get_children().await?.clone_value())
+        }
+
+        let mut routes = this
+            .page_roots
+            .iter()
+            .copied()
+            .try_flat_map_recursive_join(get_content_source_children)
+            .await?
+            .into_iter()
+            .map(content_source_to_pathname)
+            .try_join()
+            .await?
+            .into_iter()
+            .flatten()
+            .collect::<Vec<_>>();
+
         routes.sort();
 
-        Ok(StringsVc::cell(routes.into_iter().collect()))
+        Ok(StringsVc::cell(routes))
     }
 }
 

crates/next-core/src/next_font_google/mod.rs

@@ -168,7 +168,7 @@ impl ImportMappingReplacement for NextFontGoogleCssModuleReplacer {
             Ok(r) => Some(
                 update_stylesheet(r.await?.body.to_string(), options, scoped_font_family)
                     .await?
-                    .clone(),
+                    .clone_value(),
             ),
             Err(err) => {
                 // Inform the user of the failure to retreive the stylesheet, but don't