Repository: vercel/next.js
Pull request: Turbopack: Fix some eventually consistent reads at the top level in dev-server and snapshot tests

This fixes

```
UPDATE=1 cargo test -p turbopack-tests
```

when a snapshot test output is modified or deleted.  
  
We must get the `Effects` inside of an operation, as that involves reading `Vc`s, but then we should apply them outside of the operation (to guarantee that they only run once). The Next.js code follows a similar pattern here.

Changed files:
filename: turbopack/crates/turbo-tasks/src/effect.rs
patch: @@ -16,12 +16,9 @@ use tokio::task_local;
 use tracing::Instrument;
 
 use crate::{
-    self as turbo_tasks, CollectiblesSource, NonLocalValue, ReadRef, ResolvedVc, TryJoinIterExt,
-    debug::ValueDebugFormat,
-    emit,
+    self as turbo_tasks, CollectiblesSource, ReadRef, ResolvedVc, TryJoinIterExt, emit,
     event::{Event, EventListener},
     spawn,
-    trace::TraceRawVcs,
     util::SharedError,
 };
 
@@ -214,7 +211,8 @@ pub async fn get_effects(source: impl CollectiblesSource) -> Result<Effects> {
 
 /// Captured effects from an operation. This struct can be used to return Effects from a turbo-tasks
 /// function and apply them later.
-#[derive(TraceRawVcs, Default, ValueDebugFormat, NonLocalValue)]
+#[derive(Default)]
+#[turbo_tasks::value(shared, eq = "manual", serialization = "none")]
 pub struct Effects {
     #[turbo_tasks(trace_ignore, debug_ignore)]
     effects: Vec<ReadRef<EffectInstance>>,

filename: turbopack/crates/turbo-tasks/src/vc/operation.rs
patch: @@ -7,8 +7,9 @@ use serde::{Deserialize, Serialize};
 pub use turbo_tasks_macros::OperationValue;
 
 use crate::{
-    CollectiblesSource, RawVc, ReadVcFuture, ResolvedVc, TaskInput, UpcastStrict, Vc, VcValueTrait,
-    VcValueType, marker_trait::impl_auto_marker_trait, trace::TraceRawVcs,
+    CollectiblesSource, IntoTraitRef, RawVc, ReadVcFuture, ResolvedVc, TaskInput, TraitRef,
+    UpcastStrict, Vc, VcValueTrait, VcValueType, marker_trait::impl_auto_marker_trait,
+    trace::TraceRawVcs,
 };
 
 /// A "subtype" (can be converted via [`.connect()`]) of [`Vc`] that
@@ -140,6 +141,17 @@ impl<T: ?Sized> OperationVc<T> {
             .strongly_consistent()
             .into()
     }
+
+    /// [Connects the `OperationVc`][Self::connect] and returns a [strongly
+    /// consistent][crate::ReadConsistency::Strong] read of the value.
+    ///
+    /// This ensures that all internal tasks are finished before the read is returned.
+    pub async fn read_trait_strongly_consistent(self) -> Result<TraitRef<T>>
+    where
+        T: VcValueTrait,
+    {
+        self.connect().into_trait_ref().strongly_consistent().await
+    }
 }
 
 impl<T> Copy for OperationVc<T> where T: ?Sized {}

filename: turbopack/crates/turbopack-dev-server/src/lib.rs
patch: @@ -32,7 +32,7 @@ use socket2::{Domain, Protocol, Socket, Type};
 use tokio::task::JoinHandle;
 use tracing::{Instrument, Level, Span, event, info_span};
 use turbo_tasks::{
-    NonLocalValue, OperationVc, PrettyPrintError, TurboTasksApi, Vc, apply_effects,
+    Effects, NonLocalValue, OperationVc, PrettyPrintError, TurboTasksApi, Vc, get_effects,
     run_once_with_reason, trace::TraceRawVcs, util::FormatDuration,
 };
 use turbopack_core::issue::{IssueReporter, IssueSeverity, handle_issues};
@@ -57,6 +57,21 @@ where
     }
 }
 
+#[turbo_tasks::value(serialization = "none")]
+struct ContentSourceWithIssues {
+    source_op: OperationVc<Box<dyn ContentSource>>,
+    effects: Effects,
+}
+
+#[turbo_tasks::function(operation)]
+async fn get_source_with_issues_operation(
+    source_op: OperationVc<Box<dyn ContentSource>>,
+) -> Result<Vc<ContentSourceWithIssues>> {
+    let _ = source_op.resolve_strongly_consistent().await?;
+    let effects = get_effects(source_op).await?;
+    Ok(ContentSourceWithIssues { source_op, effects }.cell())
+}
+
 #[derive(TraceRawVcs, Debug, NonLocalValue)]
 pub struct DevServerBuilder {
     #[turbo_tasks(trace_ignore)]
@@ -208,12 +223,13 @@ impl DevServerBuilder {
 
                             let uri = request.uri();
                             let path = uri.path().to_string();
-                            let source_op = source_provider.get_source();
-                            // HACK: Resolve `source` now so that we can get any issues on it
-                            let _ = source_op.resolve_strongly_consistent().await?;
-                            apply_effects(source_op).await?;
+                            let source_with_issues_op =
+                                get_source_with_issues_operation(source_provider.get_source());
+                            let ContentSourceWithIssues { source_op, effects } =
+                                &*source_with_issues_op.read_strongly_consistent().await?;
+                            effects.apply().await?;
                             handle_issues(
-                                source_op,
+                                source_with_issues_op,
                                 issue_reporter,
                                 IssueSeverity::Fatal,
                                 Some(&path),
@@ -229,7 +245,7 @@ impl DevServerBuilder {
                                     // It's unlikely (the calls happen one-after-another), but this
                                     // could cause inconsistency between the reported issues and
                                     // the generated HTTP response.
-                                    source_op,
+                                    *source_op,
                                     request,
                                     issue_reporter,
                                 )

filename: turbopack/crates/turbopack-tests/tests/snapshot.rs
patch: @@ -11,7 +11,7 @@ use rustc_hash::FxHashSet;
 use serde::Deserialize;
 use serde_json::json;
 use turbo_rcstr::{RcStr, rcstr};
-use turbo_tasks::{ResolvedVc, TurboTasks, ValueToString, Vc, apply_effects};
+use turbo_tasks::{Effects, ResolvedVc, TurboTasks, ValueToString, Vc, get_effects};
 use turbo_tasks_backend::{BackendOptions, TurboTasksBackend, noop_backing_storage};
 use turbo_tasks_env::DotenvProcessEnv;
 use turbo_tasks_fs::{
@@ -222,30 +222,31 @@ async fn run(resource: PathBuf) -> Result<()> {
         noop_backing_storage(),
     ));
     tt.run_once(async move {
-        let emit_op = run_inner_operation(resource.to_str().unwrap().into());
-        emit_op.read_strongly_consistent().await?;
-        apply_effects(emit_op).await?;
+        #[turbo_tasks::function(operation)]
+        async fn inner_operation(resource: RcStr) -> Result<Vc<Effects>> {
+            let out_op = run_test_operation(resource);
+            let out_vc = out_op.resolve_strongly_consistent().await?.owned().await?;
 
-        Ok(())
-    })
-    .await?;
+            let plain_issues = out_op
+                .peek_issues()
+                .get_plain_issues(IssueFilter::everything())
+                .await?;
 
-    Ok(())
-}
-
-#[turbo_tasks::function(operation)]
-async fn run_inner_operation(resource: RcStr) -> Result<()> {
-    let out_op = run_test_operation(resource);
-    let out_vc = out_op.resolve_strongly_consistent().await?.owned().await?;
+            snapshot_issues(plain_issues, out_vc.join("issues")?, &REPO_ROOT)
+                .await
+                .context("Unable to handle issues")?;
 
-    let plain_issues = out_op
-        .peek_issues()
-        .get_plain_issues(IssueFilter::everything())
-        .await?;
+            Ok(get_effects(out_op).await?.cell())
+        }
+        inner_operation(resource.to_str().unwrap().into())
+            .read_strongly_consistent()
+            .await?
+            .apply()
+            .await?;
 
-    snapshot_issues(plain_issues, out_vc.join("issues")?, &REPO_ROOT)
-        .await
-        .context("Unable to handle issues")?;
+        Ok(())
+    })
+    .await?;
 
     Ok(())
 }