[Turbopack] gracefully stop turbo-tasks to allow persisting to comple…

…te (#69661)

### What?

Make sure to shutdown turbo-tasks gracefully to allow persisting to
complete.

crates/napi/src/next_api/project.rs

@@ -432,6 +432,13 @@ pub async fn project_update(
     Ok(())
 }
 
+#[napi(ts_return_type = "{ __napiType: \"Project\" }")]
+pub async fn project_shutdown(
+    #[napi(ts_arg_type = "{ __napiType: \"Project\" }")] project: External<ProjectInstance>,
+) {
+    project.turbo_tasks.stop_and_wait().await;
+}
+
 #[napi(object)]
 #[derive(Default)]
 struct AppPageNapiRoute {

packages/next/src/build/index.ts

@@ -1249,6 +1249,7 @@ export default async function build(
       async function turbopackBuild(): Promise<{
         duration: number
         buildTraceContext: undefined
+        shutdownPromise: Promise<void>
       }> {
         if (!IS_TURBOPACK_BUILD) {
           throw new Error("next build doesn't support turbopack yet")
@@ -1464,6 +1465,8 @@ export default async function build(
           }
         }
 
+        const shutdownPromise = project.shutdown()
+
         if (warnings.length > 0) {
           Log.warn(
             `Turbopack build collected ${warnings.length} warnings:\n${warnings
@@ -1487,6 +1490,7 @@ export default async function build(
         return {
           duration: process.hrtime(startTime)[0],
           buildTraceContext: undefined,
+          shutdownPromise,
         }
       }
 
@@ -1531,9 +1535,15 @@ export default async function build(
         },
       })
 
+      let shutdownPromise = Promise.resolve()
       if (!isGenerateMode) {
         if (turboNextBuild) {
-          const { duration: compilerDuration, ...rest } = await turbopackBuild()
+          const {
+            duration: compilerDuration,
+            shutdownPromise: p,
+            ...rest
+          } = await turbopackBuild()
+          shutdownPromise = p
           traceMemoryUsage('Finished build', nextBuildSpan)
 
           buildTraceContext = rest.buildTraceContext
@@ -3566,6 +3576,8 @@ export default async function build(
       await nextBuildSpan
         .traceChild('telemetry-flush')
         .traceAsyncFn(() => telemetry.flush())
+
+      await shutdownPromise
     })
   } finally {
     // Ensure we wait for lockfile patching if present

packages/next/src/build/swc/index.ts

@@ -679,6 +679,8 @@ export interface Project {
     aggregationMs: number
   ): AsyncIterableIterator<TurbopackResult<UpdateMessage>>
 
+  shutdown(): Promise<void>
+
   onExit(): Promise<void>
 }
 
@@ -1100,6 +1102,10 @@ function bindingToApi(
       return subscription
     }
 
+    shutdown(): Promise<void> {
+      return binding.projectShutdown(this._nativeProject)
+    }
+
     onExit(): Promise<void> {
       return binding.projectOnExit(this._nativeProject)
     }

turbopack/crates/turbo-tasks-testing/src/lib.rs

@@ -315,6 +315,10 @@ impl TurboTasksApi for VcStorage {
     ) -> std::pin::Pin<Box<dyn Future<Output = Result<()>> + Send + 'static>> {
         unimplemented!()
     }
+
+    fn stop_and_wait(&self) -> std::pin::Pin<Box<dyn Future<Output = ()> + Send + 'static>> {
+        Box::pin(async {})
+    }
 }
 
 impl VcStorage {

turbopack/crates/turbo-tasks-testing/src/run.rs

@@ -97,11 +97,13 @@ where
     println!("Run #1 (without cache)");
     let first = run_once(tt.clone(), fut()).await?;
     println!("Run #2 (with memory cache, same TurboTasks instance)");
-    let second = run_once(tt, fut()).await?;
+    let second = run_once(tt.clone(), fut()).await?;
     assert_eq!(first, second);
+    tt.stop_and_wait().await;
     let tt = registration.create_turbo_tasks(&name, false);
     println!("Run #3 (with persistent cache if available, new TurboTasks instance)");
     let third = run_once(tt.clone(), fut()).await?;
+    tt.stop_and_wait().await;
     assert_eq!(first, third);
     Ok(())
 }

turbopack/crates/turbo-tasks/src/manager.rs

@@ -191,6 +191,8 @@ pub trait TurboTasksApi: TurboTasksCallApi + Sync + Send {
         &self,
         f: Pin<Box<dyn Future<Output = Result<()>> + Send + 'static>>,
     ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'static>>;
+
+    fn stop_and_wait(&self) -> Pin<Box<dyn Future<Output = ()> + Send>>;
 }
 
 /// A wrapper around a value that is unused.
@@ -1415,6 +1417,13 @@ impl<B: Backend + 'static> TurboTasksApi for TurboTasks<B> {
             ),
         ))
     }
+
+    fn stop_and_wait(&self) -> Pin<Box<dyn Future<Output = ()> + Send + 'static>> {
+        let this = self.pin();
+        Box::pin(async move {
+            this.stop_and_wait().await;
+        })
+    }
 }
 
 impl<B: Backend + 'static> TurboTasksBackendApi<B> for TurboTasks<B> {
@@ -1431,6 +1440,7 @@ impl<B: Backend + 'static> TurboTasksBackendApi<B> for TurboTasks<B> {
             this.backend.run_backend_job(id, &*this).await;
         })
     }
+
     #[track_caller]
     fn schedule_backend_foreground_job(&self, id: BackendJobId) {
         self.schedule_foreground_job(move |this| async move {