Turbopack: aggregate server HMR into one subscription

Replace per-chunk Server HMR fan-out with a single firehose subscription that diffs every HMR-eligible chunk under the target root in one tick.  This significantly cuts the number of tokio task churn on projects with many server chunks and centralizes the diff/clear logic. It leads to a multi-second saving in both cold and warm builds in a large app.

A following PR will bring this to client chunks.

Rust:
- New `aggregate_hmr` module: `AggregateHmrVersion` keyed by chunk path, `merged_partial_update` builder, and `is_hmr_eligible_chunk` (excludes `.map` files, which would force every diff to `Total`).
- `Project::all_hmr_version_state` / `all_hmr_update` aggregate over the whole `hmr_root_path`. The seed transition emits an empty `Partial` so the JS consumer doesn't treat it as a restart and wipe handlers the triggering request just populated. Any chunk requiring `Total`/`Missing`
  escalates the batch.
- `VersionedContentMap::hmr_chunks_in_path` lists eligible chunks with their `VersionedContent`.

NAPI:
- `projectAllHmrEvents(target)` returns a single subscription.

JS:
- `setupServerHmr` subscribes once via `allHmrEvents` instead of fanning out over `hmrChunkNamesSubscribe`.
- `clear()` evicts every chunk under `server/chunks/` from `require.cache` directly rather than tracking subscriptions. This is a bit fragile as it relies on the path prefix and scanning require.cache.

Author: wbinnssmith

crates/next-api/src/aggregate_hmr.rs

@@ -0,0 +1,161 @@
+//! Aggregated HMR: one [`VersionState`] covering every chunk under a target's
+//! root, so the dev server can subscribe once instead of per chunk.
+//!
+//! [`VersionState`]: turbopack_core::version::VersionState
+
+use std::sync::Arc;
+
+use anyhow::Result;
+use rustc_hash::FxHashMap;
+use turbo_rcstr::RcStr;
+use turbo_tasks::{FxIndexMap, ResolvedVc, TraitRef, TryJoinIterExt, Vc};
+use turbo_tasks_fs::FileSystemPath;
+use turbo_tasks_hash::{Xxh3Hash64Hasher, encode_base64};
+use turbopack_core::version::{NotFoundVersion, PartialUpdate, Update, Version, VersionedContent};
+
+use crate::versioned_content_map::VersionedContentMap;
+
+/// One chunk's contribution to an [`AggregateHmrVersion`]: its output path and
+/// the versioned content backing it.
+pub struct HmrChunkWithContent {
+    pub path: RcStr,
+    pub content: ResolvedVc<Box<dyn VersionedContent>>,
+}
+
+/// Whether an emitted chunk participates in HMR. Source map (`.map`) files do
+/// not: their content fully rewrites on any source change, which would force
+/// per-chunk diffs to escalate to `Total`.
+pub fn is_hmr_eligible_chunk(name: &str) -> bool {
+    !name.ends_with(".map")
+}
+
+/// Per-chunk versions keyed by path. `id()` hashes sorted entries so it's
+/// stable across `FxIndexMap` iteration order. Mirrors `EcmascriptDevChunkListVersion`.
+#[turbo_tasks::value(serialization = "skip", shared)]
+pub struct AggregateHmrVersion {
+    #[turbo_tasks(trace_ignore)]
+    pub versions: FxIndexMap<RcStr, TraitRef<Box<dyn Version>>>,
+}
+
+#[turbo_tasks::value_impl]
+impl Version for AggregateHmrVersion {
+    #[turbo_tasks::function]
+    async fn id(&self) -> Result<Vc<RcStr>> {
+        let mut entries = self
+            .versions
+            .iter()
+            .map(|(path, version)| {
+                let path = path.clone();
+                let version = TraitRef::cell(version.clone());
+                async move {
+                    let id = version.id().owned().await?;
+                    Ok::<_, anyhow::Error>((path, id))
+                }
+            })
+            .try_join()
+            .await?;
+        entries.sort_by(|a, b| a.0.cmp(&b.0));
+
+        let mut hasher = Xxh3Hash64Hasher::new();
+        hasher.write_value(entries.len());
+        for (path, id) in entries {
+            hasher.write_value(path.as_str());
+            hasher.write_value(id.as_str());
+        }
+        Ok(Vc::cell(encode_base64(hasher.finish()).into()))
+    }
+}
+
+impl AggregateHmrVersion {
+    /// Snapshots every HMR-eligible chunk under `root` in `map` into a new
+    /// [`Version`]. Returns a [`NotFoundVersion`] when no chunks exist yet
+    /// (e.g. before any endpoints have been written).
+    pub async fn from_map(
+        map: Vc<VersionedContentMap>,
+        root: &FileSystemPath,
+    ) -> Result<Vc<Box<dyn Version>>> {
+        let chunks = map.hmr_chunks_in_path(root).await?;
+        if chunks.is_empty() {
+            return Ok(Vc::upcast(NotFoundVersion::new()));
+        }
+        Ok(Vc::upcast(Self::from_chunks(&chunks).await?))
+    }
+
+    /// Snapshots each [`HmrChunkWithContent`]'s [`Version`] into a new
+    /// [`AggregateHmrVersion`].
+    pub async fn from_chunks(chunks: &[HmrChunkWithContent]) -> Result<Vc<Self>> {
+        let versions = chunks
+            .iter()
+            .map(|HmrChunkWithContent { path, content }| {
+                let path = path.clone();
+                let content = *content;
+                async move {
+                    let version = content.version().into_trait_ref().await?;
+                    Ok::<_, anyhow::Error>((path, version))
+                }
+            })
+            .try_join()
+            .await?
+            .into_iter()
+            .collect();
+        Ok(Self { versions }.cell())
+    }
+}
+
+/// Unions one chunk's `EcmascriptMergedUpdate` into the combined `{entries, chunks}`.
+/// Both maps are keyed by globally-unique ids, so plain insertion is safe.
+pub fn merge_ecmascript_merged_update(
+    combined_entries: &mut FxHashMap<String, serde_json::Value>,
+    combined_chunks: &mut FxHashMap<String, serde_json::Value>,
+    instruction: &serde_json::Value,
+) {
+    let Some(obj) = instruction.as_object() else {
+        return;
+    };
+    if let Some(entries) = obj.get("entries").and_then(|v| v.as_object()) {
+        for (k, v) in entries {
+            combined_entries.insert(k.clone(), v.clone());
+        }
+    }
+    if let Some(chunks) = obj.get("chunks").and_then(|v| v.as_object()) {
+        for (k, v) in chunks {
+            combined_chunks.insert(k.clone(), v.clone());
+        }
+    }
+}
+
+/// Builds an `Update::Partial` whose instruction is a combined
+/// `EcmascriptMergedUpdate` covering `entries` and `chunks`. Empty maps are
+/// omitted so an empty `entries`/`chunks` field never appears in the payload.
+///
+/// Passing empty maps produces an instruction with only `type:
+/// "EcmascriptMergedUpdate"`, used to advance `VersionState` to `to` without
+/// the JS consumer applying anything: it sees a `partial` event with nothing
+/// to apply and short-circuits.
+pub fn merged_partial_update(
+    to: TraitRef<Box<dyn Version>>,
+    entries: FxHashMap<String, serde_json::Value>,
+    chunks: FxHashMap<String, serde_json::Value>,
+) -> Update {
+    let mut instruction = serde_json::Map::new();
+    instruction.insert(
+        "type".to_string(),
+        serde_json::Value::String("EcmascriptMergedUpdate".to_string()),
+    );
+    if !entries.is_empty() {
+        instruction.insert(
+            "entries".to_string(),
+            serde_json::Value::Object(entries.into_iter().collect()),
+        );
+    }
+    if !chunks.is_empty() {
+        instruction.insert(
+            "chunks".to_string(),
+            serde_json::Value::Object(chunks.into_iter().collect()),
+        );
+    }
+    Update::Partial(PartialUpdate {
+        to,
+        instruction: Arc::new(serde_json::Value::Object(instruction)),
+    })
+}

crates/next-api/src/lib.rs

@@ -2,6 +2,7 @@
 #![feature(arbitrary_self_types_pointers)]
 #![feature(impl_trait_in_assoc_type)]
 
+mod aggregate_hmr;
 pub mod analyze;
 mod app;
 mod asset_hashes_manifest;

crates/next-api/src/project.rs

@@ -37,7 +37,7 @@ use tracing::{Instrument, field::Empty};
 use turbo_rcstr::{RcStr, rcstr};
 use turbo_tasks::{
     Completion, Completions, FxIndexMap, NonLocalValue, OperationValue, OperationVc, ReadRef,
-    ResolvedVc, State, TransientInstance, TryFlatJoinIterExt, TryJoinIterExt, Vc,
+    ResolvedVc, State, TraitRef, TransientInstance, TryFlatJoinIterExt, TryJoinIterExt, Vc,
     debug::ValueDebugFormat, fxindexmap, trace::TraceRawVcs,
 };
 use turbo_tasks_env::{EnvMap, ProcessEnv};
@@ -80,7 +80,8 @@ use turbopack_core::{
     reference_type::{CommonJsReferenceSubType, ReferenceType},
     resolve::{FindContextFileResult, find_context_file},
     version::{
-        NotFoundVersion, OptionVersionedContent, Update, Version, VersionState, VersionedContent,
+        NotFoundVersion, OptionVersionedContent, PartialUpdate, TotalUpdate, Update, Version,
+        VersionState, VersionedContent,
     },
 };
 #[cfg(feature = "process_pool")]
@@ -91,6 +92,10 @@ use turbopack_node::worker_threads_backend;
 use turbopack_nodejs::NodeJsChunkingContext;
 
 use crate::{
+    aggregate_hmr::{
+        AggregateHmrVersion, HmrChunkWithContent, is_hmr_eligible_chunk,
+        merge_ecmascript_merged_update, merged_partial_update,
+    },
     app::{AppProject, OptionAppProject},
     empty::EmptyEndpoint,
     entrypoints::Entrypoints,
@@ -2498,14 +2503,158 @@ impl Project {
         }
     }
 
+    /// Aggregate counterpart to [`Self::hmr_version_state`]: one [`VersionState`]
+    /// covering every HMR-eligible chunk under `target`'s root. See
+    /// [`Self::all_hmr_update`].
+    #[turbo_tasks::function]
+    pub async fn all_hmr_version_state(
+        self: ResolvedVc<Self>,
+        target: HmrTarget,
+        session: TransientInstance<()>,
+    ) -> Result<Vc<VersionState>> {
+        if target == HmrTarget::Client {
+            bail!("all_hmr_version_state is not yet implemented for the client target");
+        }
+
+        // The session argument keeps this from caching across sessions.
+        let _ = session;
+
+        #[tracing::instrument(
+            level = "info",
+            name = "get aggregate HMR version",
+            skip_all,
+            fields(target = %target),
+        )]
+        #[turbo_tasks::function(operation, root)]
+        async fn aggregate_hmr_version_operation(
+            this: ResolvedVc<Project>,
+            target: HmrTarget,
+        ) -> Result<Vc<Box<dyn Version>>> {
+            let Some(map) = this.await?.versioned_content_map else {
+                bail!("must be in dev mode to hmr")
+            };
+            let root = this.hmr_root_path(target).owned().await?;
+            AggregateHmrVersion::from_map(*map, &root).await
+        }
+        let version_op = aggregate_hmr_version_operation(self, target);
+
+        // INVALIDATION: untracked initial read; the subscription drives invalidation.
+        let state = VersionState::new(
+            version_op
+                .read_trait_strongly_consistent()
+                .untracked()
+                .await?,
+        )
+        .await?;
+        Ok(state)
+    }
+
+    /// Aggregate counterpart to [`Self::hmr_update`]: a single `Update` whose
+    /// `EcmascriptMergedUpdate` is the union of per-chunk diffs under
+    /// `target`'s root.
+    ///
+    /// All-or-nothing restart: any chunk needing `Total`/`Missing` escalates
+    /// the whole batch to `Total` (the runtime can't partially restart). New
+    /// chunks absent from `from` are skipped; the runtime require()s them on
+    /// demand.
+    #[turbo_tasks::function]
+    pub async fn all_hmr_update(
+        self: Vc<Self>,
+        target: HmrTarget,
+        from: Vc<VersionState>,
+    ) -> Result<Vc<Update>> {
+        if target == HmrTarget::Client {
+            bail!("all_hmr_update is not yet implemented for the client target");
+        }
+
+        let Some(map) = self.await?.versioned_content_map else {
+            bail!("must be in dev mode to hmr")
+        };
+        let root = self.hmr_root_path(target).owned().await?;
+        let chunks_versioned_content = map.hmr_chunks_in_path(&root).await?;
+
+        // No chunks to diff yet (e.g. before any endpoints have been written).
+        if chunks_versioned_content.is_empty() {
+            return Ok(Update::None.cell());
+        }
+
+        // Build `to` up front so we can return it on every escape hatch below.
+        let to_aggregate = AggregateHmrVersion::from_chunks(&chunks_versioned_content).await?;
+        let to_ref = Vc::upcast::<Box<dyn Version>>(to_aggregate)
+            .into_trait_ref()
+            .await?;
+
+        let from_resolved = from.get().to_resolved().await?;
+        let Some(from_aggregate) =
+            ResolvedVc::try_downcast_type::<AggregateHmrVersion>(from_resolved)
+        else {
+            // If `from` is `None`, the last state was the initial state. In this case, emit an
+            // empty Partial update to advance the state to `to` without triggering a restart
+            return Ok(
+                merged_partial_update(to_ref, FxHashMap::default(), FxHashMap::default()).cell(),
+            );
+        };
+        let from_aggregate = from_aggregate.await?;
+
+        // Diff each chunk that exists in `from` against its current version.
+        let mut has_new_chunks = false;
+        let chunk_updates = chunks_versioned_content
+            .into_iter()
+            .filter_map(|HmrChunkWithContent { path, content }| {
+                let Some(prev) = from_aggregate.versions.get(&path).cloned() else {
+                    has_new_chunks = true;
+                    return None;
+                };
+                Some((path, content, TraitRef::cell(prev)))
+            })
+            .map(|(path, content, prev)| async move {
+                let update = content.update(prev).await?;
+                Ok::<_, anyhow::Error>((path, update))
+            })
+            .try_join()
+            .await?;
+
+        let mut combined_entries: FxHashMap<String, serde_json::Value> = FxHashMap::default();
+        let mut combined_chunks: FxHashMap<String, serde_json::Value> = FxHashMap::default();
+        for (_path, update) in chunk_updates {
+            match &*update {
+                Update::None => {}
+                Update::Missing | Update::Total(_) => {
+                    return Ok(Update::Total(TotalUpdate { to: to_ref }).cell());
+                }
+                Update::Partial(PartialUpdate { instruction, .. }) => {
+                    merge_ecmascript_merged_update(
+                        &mut combined_entries,
+                        &mut combined_chunks,
+                        instruction,
+                    );
+                }
+            }
+        }
+
+        if combined_entries.is_empty() && combined_chunks.is_empty() && !has_new_chunks {
+            return Ok(Update::None.cell());
+        }
+
+        Ok(merged_partial_update(to_ref, combined_entries, combined_chunks).cell())
+    }
+
     /// Gets a list of all HMR chunk names that can be subscribed to for the
     /// specified target. Used by the dev server to set up server-side HMR
     /// subscriptions for all Node.js App Router entries (pages and route
-    /// handlers).
+    /// handlers). See [`is_hmr_eligible_chunk`] for the eligibility rule.
     #[turbo_tasks::function]
     pub async fn hmr_chunk_names(self: Vc<Self>, target: HmrTarget) -> Result<Vc<Vec<RcStr>>> {
         if let Some(map) = self.await?.versioned_content_map {
-            Ok(map.keys_in_path(self.hmr_root_path(target).owned().await?))
+            let names = map
+                .keys_in_path(self.hmr_root_path(target).owned().await?)
+                .await?;
+            let filtered: Vec<RcStr> = names
+                .iter()
+                .filter(|name| is_hmr_eligible_chunk(name))
+                .cloned()
+                .collect();
+            Ok(Vc::cell(filtered))
         } else {
             bail!("must be in dev mode to hmr")
         }