dev-server: avoid top-level eventually-consistent reads

Under release-with-assertions (debug-assertions on), the dev server tripped
debug_assert_not_in_top_level_task by reading task outputs eventually
consistently from top-level run_once tasks, panicking worker threads and
hanging the turbopack-cli benchmark.

- Make IssueReporter::report_issues a plain async method (not a turbo-task)
  that takes pre-collected PlainIssues; collect them in a new collect_issues
  operation read strongly consistently in handle_issues. Mark ConsoleUi::new
  root so the reporter can be resolved strongly consistently.
- Apply content-source side effects via a new apply_side_effects_operation
  read strongly consistently instead of awaiting each apply() at the top level.
- Compute the initial HMR version in initial_version_operation and read the
  update-stream content strongly consistently in UpdateStream::new.

Author: lukesandberg

Cargo.lock

@@ -16,6 +16,7 @@ workspace = true
 
 [dependencies]
 anyhow = { workspace = true }
+async-trait = { workspace = true }
 crossterm = "0.26.0"
 owo-colors = { workspace = true }
 rustc-hash = { workspace = true }

turbopack/crates/turbopack-cli-utils/Cargo.toml

@@ -8,15 +8,16 @@ use std::{
 };
 
 use anyhow::Result;
+use async_trait::async_trait;
 use crossterm::style::{StyledContent, Stylize};
 use owo_colors::{OwoColorize as _, Style};
 use rustc_hash::{FxHashMap, FxHashSet};
 use turbo_rcstr::RcStr;
-use turbo_tasks::{RawVc, TransientInstance, TransientValue, Vc};
+use turbo_tasks::{RawVc, ReadRef, TransientInstance, Vc};
 use turbo_tasks_fs::{FileLinesContent, source_context::get_source_context};
 use turbopack_core::issue::{
-    CollectibleIssuesExt, IssueFilter, IssueReporter, IssueSeverity, PlainIssue, PlainIssueSource,
-    PlainTraceItem, StyledString,
+    IssueReporter, IssueSeverity, PlainIssue, PlainIssueSource, PlainIssues, PlainTraceItem,
+    StyledString,
 };
 
 use crate::source_context::format_source_context_lines;
@@ -366,7 +367,7 @@ impl PartialEq for ConsoleUi {
 
 #[turbo_tasks::value_impl]
 impl ConsoleUi {
-    #[turbo_tasks::function]
+    #[turbo_tasks::function(root)]
     pub fn new(options: TransientInstance<LogOptions>) -> Vc<Self> {
         ConsoleUi {
             options: (*options).clone(),
@@ -376,15 +377,15 @@ impl ConsoleUi {
     }
 }
 
+#[async_trait]
 #[turbo_tasks::value_impl]
 impl IssueReporter for ConsoleUi {
-    #[turbo_tasks::function]
     async fn report_issues(
         &self,
-        source: TransientValue<RawVc>,
+        issues: ReadRef<PlainIssues>,
+        source: RawVc,
         min_failing_severity: IssueSeverity,
-    ) -> Result<Vc<bool>> {
-        let issues = source.peek_issues();
+    ) -> Result<bool> {
         let LogOptions {
             ref current_dir,
             ref project_dir,
@@ -395,7 +396,7 @@ impl IssueReporter for ConsoleUi {
         } = self.options;
         let mut grouped_issues: GroupedIssues = FxHashMap::default();
 
-        let plain_issues = issues.get_plain_issues(&IssueFilter::everything()).await?;
+        let plain_issues = &issues.0;
         let issues = plain_issues
             .iter()
             .map(|plain_issue| {
@@ -405,11 +406,7 @@ impl IssueReporter for ConsoleUi {
             .collect::<Vec<_>>();
 
         let issue_ids = issues.iter().map(|(_, id)| *id).collect::<FxHashSet<_>>();
-        let mut new_ids = self
-            .seen
-            .lock()
-            .unwrap()
-            .new_ids(source.into_value(), issue_ids);
+        let mut new_ids = self.seen.lock().unwrap().new_ids(source, issue_ids);
 
         let mut has_fatal = false;
         for (plain_issue, id) in issues {
@@ -539,7 +536,7 @@ impl IssueReporter for ConsoleUi {
             }
         }
 
-        Ok(Vc::cell(has_fatal))
+        Ok(has_fatal)
     }
 }
 

turbopack/crates/turbopack-cli-utils/src/issue.rs

@@ -16,9 +16,9 @@ use serde::{Deserialize, Serialize};
 use turbo_esregex::EsRegex;
 use turbo_rcstr::{RcStr, rcstr};
 use turbo_tasks::{
-    CollectiblesSource, NonLocalValue, OperationVc, RawVc, ReadRef, ResolvedVc, TransientValue,
-    TryFlatJoinIterExt, TryJoinIterExt, Upcast, ValueDefault, ValueToString, ValueToStringRef, Vc,
-    emit, trace::TraceRawVcs,
+    CollectiblesSource, NonLocalValue, OperationVc, RawVc, ReadRef, ResolvedVc, TryFlatJoinIterExt,
+    TryJoinIterExt, Upcast, ValueDefault, ValueToString, ValueToStringRef, Vc, emit,
+    trace::TraceRawVcs,
 };
 use turbo_tasks_fs::{
     FileContent, FileLine, FileLinesContent, FileSystem, FileSystemPath, glob::Glob,
@@ -904,6 +904,14 @@ pub struct PlainIssue {
     pub import_traces: Vec<PlainTrace>,
 }
 
+/// A collection of [`PlainIssue`]s collected from a single source.
+///
+/// Returned by [`collect_issues`] so that the (plain) issues can be read strongly
+/// consistently from a top-level task and handed to a non-turbo-task [`IssueReporter`].
+#[turbo_tasks::value(serialization = "skip")]
+#[derive(Debug)]
+pub struct PlainIssues(pub Vec<ReadRef<PlainIssue>>);
+
 #[turbo_tasks::value(serialization = "skip")]
 #[derive(Clone, Debug, PartialOrd, Ord)]
 pub struct PlainAdditionalIssueSource {
@@ -1062,24 +1070,30 @@ impl PlainSource {
     }
 }
 
+#[async_trait]
 #[turbo_tasks::value_trait]
 pub trait IssueReporter {
-    /// Reports issues to the user (e.g. to stdio). Returns whether fatal
+    /// Reports already-collected issues to the user (e.g. to stdio). Returns whether fatal
     /// (program-ending) issues were present.
     ///
+    /// This is intentionally *not* a `#[turbo_tasks::function]`: it performs no turbo-tasks
+    /// reads of its own (the issues are collected ahead of time by [`collect_issues`]), so it
+    /// is safe to call from a top-level task.
+    ///
     /// # Arguments:
     ///
-    /// * `source` - The root [Vc] from which issues are traced. Can be used by implementers to
-    ///   determine which issues are new.  This must be derived from the OperationVc so issues can
-    ///   be collected.
-    /// * `min_failing_severity` - The minimum Vc<[IssueSeverity]>
-    ///  The minimum issue severity level considered to fatally end the program.
-    #[turbo_tasks::function]
-    fn report_issues(
-        self: Vc<Self>,
-        source: TransientValue<RawVc>,
+    /// * `issues` - The plain issues already collected from the source.
+    /// * `source` - The root [`RawVc`] from which the issues were traced. Can be used by
+    ///   implementers as a dedup key to determine which issues are new. This must be derived from
+    ///   the `OperationVc` the issues were collected from.
+    /// * `min_failing_severity` - The minimum issue severity level considered to fatally end the
+    ///   program.
+    async fn report_issues(
+        &self,
+        issues: ReadRef<PlainIssues>,
+        source: RawVc,
         min_failing_severity: IssueSeverity,
-    ) -> Vc<bool>;
+    ) -> Result<bool>;
 }
 
 pub trait CollectibleIssuesExt
@@ -1117,6 +1131,21 @@ where
     }
 }
 
+/// Collects all issues emitted by `source` as resolved [`PlainIssue`]s.
+///
+/// This is an `operation` function so its (plain) result can be read *strongly consistently* (via
+/// [`OperationVc::read_strongly_consistent`]) from a top-level task without tripping the
+/// eventually-consistent-read assertion. The per-issue `PlainIssue::from_issue` reads happen
+/// *inside* this task, where eventually-consistent reads are legal.
+#[turbo_tasks::function(operation, root)]
+async fn collect_issues(source: OperationVc<()>) -> Result<Vc<PlainIssues>> {
+    let plain = source
+        .peek_issues()
+        .get_plain_issues(&IssueFilter::everything())
+        .await?;
+    Ok(PlainIssues(plain).cell())
+}
+
 /// A helper function to print out issues to the console.
 ///
 /// Must be called in a turbo-task as this constructs a `cell`
@@ -1129,13 +1158,31 @@ pub async fn handle_issues<T: Send>(
 ) -> Result<()> {
     let source_vc = source_op.connect();
     let _ = source_op.resolve().strongly_consistent().await?;
+    let source_raw = Vc::into_raw(source_vc);
+
+    // Collect the issues in a dedicated `operation` task and read its *plain* result strongly
+    // consistently. This is safe at the top level (unlike an eventually-consistent read), while
+    // the per-issue reads happen inside `collect_issues`. The source is type-erased to
+    // `OperationVc<()>` so a single non-generic task can collect issues for any source.
+    let erased_source = OperationVc::<()>::try_from(source_raw)?;
+    let issues = collect_issues(erased_source)
+        .read_strongly_consistent()
+        .await?;
 
-    let has_fatal = issue_reporter.report_issues(
-        TransientValue::new(Vc::into_raw(source_vc)),
-        min_failing_severity,
-    );
+    // `report_issues` is a plain async method; reach it via a `TraitRef`. Resolve the reporter
+    // strongly consistently first so that `into_trait_ref` is a plain cell read (rather than an
+    // eventually-consistent task-output read) at the top level.
+    let reporter = issue_reporter
+        .to_resolved()
+        .strongly_consistent()
+        .await?
+        .into_trait_ref()
+        .await?;
+    let has_fatal = reporter
+        .report_issues(issues, source_raw, min_failing_severity)
+        .await?;
 
-    if *has_fatal.await? {
+    if has_fatal {
         let mut message = "Fatal issue(s) occurred".to_owned();
         if let Some(path) = path.as_ref() {
             message += &format!(" in {path}");

turbopack/crates/turbopack-core/src/issue/mod.rs

@@ -30,8 +30,8 @@ use socket2::{Domain, Protocol, Socket, Type};
 use tokio::task::JoinHandle;
 use tracing::{Instrument, Level, Span, event, info_span};
 use turbo_tasks::{
-    Effects, NonLocalValue, OperationVc, PrettyPrintError, TurboTasksApi, Vc,
-    read_strongly_consistent_and_apply_effects, run_once_with_reason, take_effects,
+    Completion, Effects, NonLocalValue, OperationVc, PrettyPrintError, ResolvedVc, TurboTasksApi,
+    Vc, read_strongly_consistent_and_apply_effects, run_once_with_reason, take_effects,
     trace::TraceRawVcs, util::FormatDuration,
 };
 use turbopack_core::issue::{IssueReporter, IssueSeverity, handle_issues};
@@ -71,6 +71,19 @@ async fn get_source_with_issues_operation(
     Ok(ContentSourceWithIssues { source_op, effects }.cell())
 }
 
+/// Applies all collected [`ContentSourceSideEffect`]s. The individual `apply()` reads happen
+/// *inside* this task (where eventually-consistent reads are legal); the caller reads the result
+/// strongly consistently so the work is finished before the response is observed.
+#[turbo_tasks::function(operation, root)]
+async fn apply_side_effects_operation(
+    side_effects: Vec<ResolvedVc<Box<dyn ContentSourceSideEffect>>>,
+) -> Result<Vc<Completion>> {
+    for side_effect in &side_effects {
+        side_effect.apply().await?;
+    }
+    Ok(Completion::new())
+}
+
 #[derive(TraceRawVcs, Debug, NonLocalValue)]
 pub struct DevServerBuilder {
     #[turbo_tasks(trace_ignore)]
@@ -266,13 +279,17 @@ impl DevServerBuilder {
                                 );
                             }
                             if !side_effects.is_empty() {
+                                let side_effects: Vec<_> = side_effects.into_iter().collect();
                                 let join_handle = tokio::spawn(run_once_with_reason(
                                     tt.clone(),
                                     side_effects_reason,
                                     async move {
-                                        for side_effect in side_effects {
-                                            side_effect.apply().await?;
-                                        }
+                                        // Apply the side effects inside a dedicated `operation`
+                                        // task and read its result strongly consistently, so this
+                                        // top-level task performs no eventually-consistent read.
+                                        apply_side_effects_operation(side_effects)
+                                            .read_strongly_consistent()
+                                            .await?;
                                         Ok(())
                                     },
                                 ));

turbopack/crates/turbopack-dev-server/src/lib.rs

@@ -111,6 +111,22 @@ fn versioned_content_update_operation(
     content.update(*from)
 }
 
+/// Computes the initial [`Version`] for an update stream from a resolved source request. Runs as
+/// an `operation` so [`UpdateStream::new`] can read it strongly consistently from its top-level
+/// task without performing an eventually-consistent read.
+#[turbo_tasks::function(operation, root)]
+async fn initial_version_operation(
+    content: OperationVc<ResolveSourceRequestResult>,
+) -> Result<Vc<Box<dyn Version>>> {
+    Ok(match *content.read_strongly_consistent().await? {
+        ResolveSourceRequestResult::Static(static_content, _) => {
+            static_content.await?.content.version()
+        }
+        ResolveSourceRequestResult::HttpProxy(proxy_result) => Vc::upcast(proxy_result.connect()),
+        _ => Vc::upcast(NotFoundVersion::new()),
+    })
+}
+
 #[turbo_tasks::function(operation, root)]
 async fn get_update_stream_item_operation(
     resource: RcStr,
@@ -277,17 +293,13 @@ impl UpdateStream {
 
         let content = get_content.call();
         // We can ignore issues reported in content here since [compute_update_stream]
-        // will handle them
-        let version = match *content.connect().await? {
-            ResolveSourceRequestResult::Static(static_content, _) => {
-                static_content.await?.content.version()
-            }
-            ResolveSourceRequestResult::HttpProxy(proxy_result) => {
-                Vc::upcast(proxy_result.connect())
-            }
-            _ => Vc::upcast(NotFoundVersion::new()),
-        };
-        let version_state = VersionState::new(version.into_trait_ref().await?).await?;
+        // will handle them. This runs in a top-level task (`UpdateServer::run`'s
+        // `start_once_process`), so the initial version is computed in a dedicated `operation`
+        // task (where the per-content reads are legal) and read strongly consistently.
+        let version = initial_version_operation(content)
+            .read_trait_strongly_consistent()
+            .await?;
+        let version_state = VersionState::new(version).await?;
 
         let _ = compute_update_stream(
             resource,