mirror of
https://github.com/hanzoai/next.js.git
synced 2026-10-30 14:46:38 +00:00
5d0dab5fc6
### What? Replaces the silent "make root node" promotion in the turbo-tasks backend with a panic that enforces tasks to already have the `root` attribute before performing strongly consistent reads, reading task collectibles, or removing collectibles. ### Why? Previously, the backend would silently promote any non-root task to a root node (aggregation number `u32::MAX`) when these operations were requested. This masked incorrect task configuration — tasks that needed root-level aggregation weren't explicitly declared as such, making it harder to reason about the aggregation graph and hiding potential performance issues from implicit promotions. ### How? **Backend enforcement (3 locations):** - **Strongly consistent read** (`backend/mod.rs`): Checks `NativeFunction.is_root` on the target task. If it's a persistent task without `root`, panics with both the target and reader task descriptions. - **Read task collectibles** (`backend/mod.rs`): Same check when reading collectibles from a task. - **Remove collectibles** (`operation/update_collectible.rs`): Same check when removing collectibles (count < 0). All three locations drop the task guard before panicking to avoid deadlocking with `debug_get_task_description`. **Macro support for `root` on methods:** - `value_impl_macro.rs`: Now reads the `root` attribute from `#[turbo_tasks::function(root)]` on inherent impl and trait impl methods (previously hardcoded to `false`). - `value_trait_macro.rs`: Same for trait default methods. **`root` attribute additions:** - All `#[turbo_tasks::function(operation)]` in test files, benchmarks, and fuzz code - Production operations in `crates/next-api/` that are read with strong consistency - Regular functions and methods in tests that use `.strongly_consistent()` - Trait methods in `value_impl` and `value_trait` blocks used with strong consistency **New test:** - `non_root_task_panic.rs`: Verifies the panic fires when attempting a strongly consistent read on a non-root operation task. Captures the panic from the worker thread via a panic hook since the panic propagates as a channel error to the test thread. <!-- NEXT_JS_LLM_PR --> --------- Co-authored-by: Claude Opus 4.7 (1M context) <[email protected]>
278 lines
9.5 KiB
Rust
278 lines
9.5 KiB
Rust
#![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(®ISTRATION, || 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(®ISTRATION, || 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(®ISTRATION, || 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(®ISTRATION, || 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(®ISTRATION, || 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(®ISTRATION, || 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(®ISTRATION, || 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, root)]
|
|
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, root)]
|
|
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, root)]
|
|
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, root)]
|
|
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, root)]
|
|
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, root)]
|
|
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(
|
|
123,
|
|
)));
|
|
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, root)]
|
|
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))
|
|
}
|