next.js/turbopack/crates/turbo-tasks-backend/tests/collectibles.rs at release-app-smoke-test/new-release-flow

mirror of https://github.com/hanzoai/next.js.git synced 2026-10-30 14:46:38 +00:00

Files

T

Benjamin WoodruffandGitHub 89f3649a00 Turbopack: Panic if a top-level task attempts an eventually consistent read (#89735 )

We've had some really bad bugs in the past, which can be hard to root-cause due to eventually-consistent reads a the top-level, e.g. https://github.com/vercel/next.js/pull/77511

Outside of tests, there's no good justification for doing an eventually-consistent read at the top level. It may leak incomplete computations, which can contain errors that would otherwise be resolved by a strongly consistent read.

The basic idea here is simple:

- Store a flag in `CURRENT_TASK_STATE` indicating that we're in a top-level task.
- When reading a task output or cell, check for that flag.

But, there are a few complications:

- `RawVc`'s read and resolution logic is wrong, and internally uses multiple eventually consistent operations after a strongly consistent read.
- For this to be correct, it needs to happen as one atomic operation inside of the task backend, but it would change the backend API contract, so it's not straightforward to do that.
- I added a `SUPPRESS_EVENTUAL_CONSISTENCY_TOP_LEVEL_TASK_CHECK` flag that I'm using for now.
- There are a ton of places that are doing eventually consistent reads for no good reason. These aren't causing problems today, but we should fix them instead of suppressing them. Claude helped with this, but it's still a lot.
- Some tests actually do need eventually-consistent reads because they're testing some internal behavior of turbo-tasks. I added an intentionally long-named `unmark_top_level_task_may_leak_eventually_consistent_state` helper for this.

I manually tested `turbopack-cli` against our bundler-benchmark repository that uses it.

2026-02-26 16:14:30 -08:00

278 lines

9.5 KiB

Rust

Raw Permalink Blame History

 #![feature(arbitrary_self_types)]
 #![feature(arbitrary_self_types_pointers)]
 #![allow(clippy::needless_return)] // tokio macro-generated code doesn't respect this
 use std::time::Duration;
 use anyhow::Result;
 use auto_hash_map::AutoSet;
 use rustc_hash::FxHashSet;
 use tokio::time::sleep;
 use turbo_rcstr::{RcStr, rcstr};
 use turbo_tasks::{
     CollectiblesSource, ResolvedVc, ValueToString, Vc, emit,
     unmark_top_level_task_may_leak_eventually_consistent_state,
 };
 use turbo_tasks_testing::{Registration, register, run_once};
 static REGISTRATION: Registration = register!();
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn test_transitive_emitting() {
     run_once(&REGISTRATION, || async {
         unmark_top_level_task_may_leak_eventually_consistent_state();
         let result_op = my_transitive_emitting_function(rcstr!(""), rcstr!(""));
         let result_val = result_op.connect().strongly_consistent().await?;
         let list = result_op.peek_collectibles::<Box<dyn ValueToString>>();
         assert_eq!(list.len(), 2);
         let mut expected = ["123", "42"].into_iter().collect::<FxHashSet<_>>();
         for collectible in list {
             assert!(expected.remove(collectible.to_string().await?.as_str()))
         }
         assert_eq!(result_val.0, 0);
         anyhow::Ok(())
     })
     .await
     .unwrap()
 }
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn test_transitive_emitting_indirect() {
     run_once(&REGISTRATION, || async {
         unmark_top_level_task_may_leak_eventually_consistent_state();
         let result_op = my_transitive_emitting_function(rcstr!(""), rcstr!(""));
         let collectibles_op = my_transitive_emitting_function_collectibles(rcstr!(""), rcstr!(""));
         let list = collectibles_op.connect().strongly_consistent().await?;
         assert_eq!(list.len(), 2);
         let mut expected = ["123", "42"].into_iter().collect::<FxHashSet<_>>();
         for collectible in list.iter() {
             assert!(expected.remove(collectible.to_string().await?.as_str()))
         }
         assert_eq!(result_op.connect().await?.0, 0);
         anyhow::Ok(())
     })
     .await
     .unwrap()
 }
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn test_multi_emitting() {
     run_once(&REGISTRATION, || async {
         unmark_top_level_task_may_leak_eventually_consistent_state();
         let result_op = my_multi_emitting_function();
         let result_val = result_op.connect().strongly_consistent().await?;
         let list = result_op.peek_collectibles::<Box<dyn ValueToString>>();
         assert_eq!(list.len(), 2);
         let mut expected = ["123", "42"].into_iter().collect::<FxHashSet<_>>();
         for collectible in list {
             assert!(expected.remove(collectible.to_string().await?.as_str()))
         }
         assert_eq!(result_val.0, 0);
         anyhow::Ok(())
     })
     .await
     .unwrap()
 }
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn taking_collectibles() {
     run_once(&REGISTRATION, || async {
         let result_op = my_collecting_function();
         let result_val = result_op.connect().strongly_consistent().await?;
         let list = result_op.take_collectibles::<Box<dyn ValueToString>>();
         // my_collecting_function already processed the collectibles so the list should
         // be empty
         assert!(list.is_empty());
         assert_eq!(result_val.0, 0);
         anyhow::Ok(())
     })
     .await
     .unwrap()
 }
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn taking_collectibles_extra_layer() {
     run_once(&REGISTRATION, || async {
         let result_op = my_collecting_function_indirect();
         let result_val = result_op.connect().strongly_consistent().await?;
         let list = result_op.take_collectibles::<Box<dyn ValueToString>>();
         // my_collecting_function already processed the collectibles so the list should
         // be empty
         assert!(list.is_empty());
         assert_eq!(result_val.0, 0);
         anyhow::Ok(())
     })
     .await
     .unwrap()
 }
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn taking_collectibles_parallel() {
     run_once(&REGISTRATION, || async {
         let result_op = my_transitive_emitting_function(rcstr!(""), rcstr!("a"));
         let result_val = result_op.connect().strongly_consistent().await?;
         let list = result_op.take_collectibles::<Box<dyn ValueToString>>();
         assert_eq!(list.len(), 2);
         assert_eq!(result_val.0, 0);
         let result_op = my_transitive_emitting_function(rcstr!(""), rcstr!("b"));
         let result_val = result_op.connect().strongly_consistent().await?;
         let list = result_op.take_collectibles::<Box<dyn ValueToString>>();
         assert_eq!(list.len(), 2);
         assert_eq!(result_val.0, 0);
         let result_op =
             my_transitive_emitting_function_with_child_scope(rcstr!(""), rcstr!("b"), rcstr!("1"));
         let result_val = result_op.connect().strongly_consistent().await?;
         let list = result_op.take_collectibles::<Box<dyn ValueToString>>();
         assert_eq!(list.len(), 2);
         assert_eq!(result_val.0, 0);
         let result_op =
             my_transitive_emitting_function_with_child_scope(rcstr!(""), rcstr!("b"), rcstr!("2"));
         let result_val = result_op.connect().strongly_consistent().await?;
         let list = result_op.take_collectibles::<Box<dyn ValueToString>>();
         assert_eq!(list.len(), 2);
         assert_eq!(result_val.0, 0);
         let result_op =
             my_transitive_emitting_function_with_child_scope(rcstr!(""), rcstr!("c"), rcstr!("3"));
         let result_val = result_op.connect().strongly_consistent().await?;
         let list = result_op.take_collectibles::<Box<dyn ValueToString>>();
         assert_eq!(list.len(), 2);
         assert_eq!(result_val.0, 0);
         anyhow::Ok(())
     })
     .await
     .unwrap()
 }
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn taking_collectibles_with_resolve() {
     run_once(&REGISTRATION, || async {
         let result_op = my_transitive_emitting_function_with_resolve(rcstr!("resolve"));
         result_op.connect().strongly_consistent().await?;
         let list = result_op.take_collectibles::<Box<dyn ValueToString>>();
         assert_eq!(list.len(), 2);
         anyhow::Ok(())
     })
     .await
     .unwrap()
 }
 #[turbo_tasks::value(transparent)]
 struct Collectibles(AutoSet<ResolvedVc<Box<dyn ValueToString>>>);
 #[turbo_tasks::function(operation)]
 async fn my_collecting_function() -> Result<Vc<Thing>> {
     let result_op = my_transitive_emitting_function(rcstr!(""), rcstr!(""));
     let result_vc = result_op.connect();
     result_vc.await?;
     result_op.drop_collectibles::<Box<dyn ValueToString>>();
     Ok(result_vc)
 }
 #[turbo_tasks::function(operation)]
 async fn my_collecting_function_indirect() -> Result<Vc<Thing>> {
     let result_op = my_collecting_function();
     let result_vc = result_op.connect();
     result_vc.strongly_consistent().await?;
     let list = result_op.peek_collectibles::<Box<dyn ValueToString>>();
     // my_collecting_function already processed the collectibles so the list should
     // be empty
     assert!(list.is_empty());
     Ok(result_vc)
 }
 #[turbo_tasks::function(operation)]
 async fn my_multi_emitting_function() -> Result<Vc<Thing>> {
     my_transitive_emitting_function(rcstr!(""), rcstr!("a"))
         .connect()
         .await?;
     my_transitive_emitting_function(rcstr!(""), rcstr!("b"))
         .connect()
         .await?;
     my_emitting_function(rcstr!("")).await?;
     Ok(Thing::cell(Thing(0)))
 }
 #[turbo_tasks::function(operation)]
 async fn my_transitive_emitting_function(key: RcStr, key2: RcStr) -> Result<Vc<Thing>> {
     let _ = key2;
     my_emitting_function(key).await?;
     Ok(Thing::cell(Thing(0)))
 }
 #[turbo_tasks::function(operation)]
 fn my_transitive_emitting_function_collectibles(
     key: RcStr,
     key2: RcStr,
 ) -> Result<Vc<Collectibles>> {
     let result_op = my_transitive_emitting_function(key, key2);
     Ok(Vc::cell(
         result_op.peek_collectibles::<Box<dyn ValueToString>>(),
     ))
 }
 #[turbo_tasks::function(operation)]
 async fn my_transitive_emitting_function_with_child_scope(
     key: RcStr,
     key2: RcStr,
     key3: RcStr,
 ) -> Result<Vc<Thing>> {
     let _ = key3;
     let thing_op = my_transitive_emitting_function(key, key2);
     let thing_vc = thing_op.connect();
     thing_vc.await?;
     let list = thing_op.peek_collectibles::<Box<dyn ValueToString>>();
     assert_eq!(list.len(), 2);
     Ok(thing_vc)
 }
 #[turbo_tasks::function]
 async fn my_emitting_function(key: RcStr) -> Result<()> {
     let _ = key;
     sleep(Duration::from_millis(100)).await;
     emit(ResolvedVc::upcast::<Box<dyn ValueToString>>(Thing::new(
 ,
     )));
     emit(ResolvedVc::upcast::<Box<dyn ValueToString>>(Thing::new(42)));
     Ok(())
 }
 #[turbo_tasks::function]
 fn my_transitive_emitting_function_with_thing(key: RcStr, _thing: Vc<Thing>) -> Result<()> {
     let _ = my_emitting_function(key);
     Ok(())
 }
 #[turbo_tasks::function(operation)]
 fn my_transitive_emitting_function_with_resolve(key: RcStr) -> Result<()> {
     let _ = my_transitive_emitting_function_with_thing(key, get_thing(0));
     Ok(())
 }
 #[turbo_tasks::value(shared)]
 struct Thing(u32);
 impl Thing {
     fn new(v: u32) -> ResolvedVc<Self> {
         Self::resolved_cell(Thing(v))
     }
 }
 #[turbo_tasks::value_impl]
 impl ValueToString for Thing {
     #[turbo_tasks::function]
     fn to_string(&self) -> Vc<RcStr> {
         Vc::cell(self.0.to_string().into())
     }
 }
 #[turbo_tasks::function]
 fn get_thing(v: u32) -> Vc<Thing> {
     Thing::cell(Thing(v))
 }

278 lines 9.5 KiB Rust Raw Permalink Blame History

278 lines

9.5 KiB

Rust

Raw Permalink Blame History