1use std::{iter, process::ExitStatus, time::Duration};
2
3use anyhow::{Result, bail};
4use async_trait::async_trait;
5use bincode::{Decode, Encode};
6use bytes::Bytes;
7use futures_retry::{FutureRetry, RetryPolicy};
8use serde::{Deserialize, Serialize, de::DeserializeOwned};
9use serde_json::Value as JsonValue;
10use turbo_rcstr::{RcStr, rcstr};
11use turbo_tasks::{
12 Completion, FxIndexMap, OperationVc, PrettyPrintError, ResolvedVc, TryJoinIterExt, Vc,
13 duration_span, fxindexmap, parallel::available_parallelism,
14 resolve_strongly_consistent_and_take_and_apply_effects, trace::TraceRawVcs,
15};
16use turbo_tasks_env::{EnvMap, ProcessEnv};
17use turbo_tasks_fs::{File, FileContent, FileSystemPath, to_sys_path};
18use turbopack_core::{
19 asset::AssetContent,
20 changed::content_changed,
21 chunk::{ChunkingContext, ChunkingContextExt, EvaluatableAsset, EvaluatableAssets},
22 context::AssetContext,
23 file_source::FileSource,
24 ident::AssetIdent,
25 issue::{Issue, IssueExt, IssueSource, IssueStage, StyledString},
26 module::Module,
27 module_graph::{
28 GraphEntries, ModuleGraph,
29 chunk_group_info::{ChunkGroup, ChunkGroupEntry, EntryHeuristics},
30 },
31 output::{OutputAsset, OutputAssets},
32 reference_type::{InnerAssets, ReferenceType},
33 source::Source,
34 virtual_source::VirtualSource,
35};
36
37use crate::{
38 AssetsForSourceMapping,
39 backend::{CreatePoolOptions, NodeBackend},
40 embed_js::embed_file_path,
41 emit, emit_package_json,
42 format::FormattingMode,
43 internal_assets_for_source_mapping,
44 pool_stats::PoolStatsSnapshot,
45 source_map::StructuredError,
46};
47
48#[derive(Serialize)]
49#[serde(tag = "type", rename_all = "camelCase")]
50enum EvalJavaScriptOutgoingMessage<'a> {
51 #[serde(rename_all = "camelCase")]
52 Evaluate { args: Vec<&'a JsonValue> },
53 Result {
54 id: u64,
55 data: Option<JsonValue>,
56 error: Option<String>,
57 },
58}
59
60#[derive(Deserialize, Debug)]
61#[serde(tag = "type", rename_all = "camelCase")]
62enum EvalJavaScriptIncomingMessage {
63 Info { data: JsonValue },
64 Request { id: u64, data: JsonValue },
65 End { data: Option<String> },
66 Error(StructuredError),
67}
68
69#[turbo_tasks::value(
70 cell = "new",
71 serialization = "skip",
72 evict = "last",
73 eq = "manual",
74 shared
75)]
76pub struct EvaluatePool {
77 #[turbo_tasks(trace_ignore, debug_ignore)]
78 pool: Box<dyn EvaluateOperation>,
79 pub assets_for_source_mapping: ResolvedVc<AssetsForSourceMapping>,
80 pub assets_root: FileSystemPath,
81 pub project_dir: FileSystemPath,
82}
83
84impl EvaluatePool {
85 pub(crate) fn new(
86 pool: Box<dyn EvaluateOperation>,
87 assets_for_source_mapping: ResolvedVc<AssetsForSourceMapping>,
88 assets_root: FileSystemPath,
89 project_dir: FileSystemPath,
90 ) -> Self {
91 Self {
92 pool,
93 assets_for_source_mapping,
94 assets_root,
95 project_dir,
96 }
97 }
98
99 pub async fn operation(&self) -> Result<Box<dyn Operation>> {
100 self.pool.operation().await
101 }
102
103 pub fn stats(&self) -> PoolStatsSnapshot {
104 self.pool.stats()
105 }
106
107 pub fn pre_warm(&self) {
108 self.pool.pre_warm()
109 }
110}
111
112#[async_trait::async_trait]
113pub trait EvaluateOperation: Send + Sync {
114 async fn operation(&self) -> Result<Box<dyn Operation>>;
115 fn stats(&self) -> PoolStatsSnapshot;
116 fn pre_warm(&self);
122}
123
124#[async_trait::async_trait]
125pub trait Operation: Send {
126 async fn recv(&mut self) -> Result<Bytes>;
127
128 async fn send(&mut self, data: Bytes) -> Result<()>;
129
130 async fn wait_or_kill(&mut self) -> Result<ExitStatus>;
131
132 fn disallow_reuse(&mut self) -> ();
133}
134
135#[turbo_tasks::value]
136struct EmittedEvaluatePoolAssets {
137 bootstrap: ResolvedVc<Box<dyn OutputAsset>>,
138 output_root: FileSystemPath,
139 entrypoint: FileSystemPath,
140}
141
142#[turbo_tasks::function(operation, root)]
143async fn emit_evaluate_pool_assets_operation(
144 entries: ResolvedVc<EvaluateEntries>,
145 chunking_context: ResolvedVc<Box<dyn ChunkingContext>>,
146 module_graph: ResolvedVc<ModuleGraph>,
147) -> Result<Vc<EmittedEvaluatePoolAssets>> {
148 let EvaluateEntries {
149 entries,
150 main_entry_ident,
151 } = &*entries.await?;
152
153 let entrypoint = chunking_context
154 .chunk_path(
155 None,
156 **main_entry_ident,
157 Some(rcstr!("pool_entry")),
158 rcstr!(".js"),
159 )
160 .owned()
161 .await?;
162
163 let bootstrap = chunking_context.root_entry_chunk_group_asset(
164 entrypoint.clone(),
165 ChunkGroup::Entry(entries.iter().cloned().map(ResolvedVc::upcast).collect()),
166 *module_graph,
167 OutputAssets::empty(),
168 OutputAssets::empty(),
169 );
170
171 let output_root = chunking_context.output_root().owned().await?;
172 emit_package_json(output_root.clone())?
173 .as_side_effect()
174 .await?;
175 emit(bootstrap, output_root.clone())
176 .as_side_effect()
177 .await?;
178
179 Ok(EmittedEvaluatePoolAssets {
180 bootstrap: bootstrap.to_resolved().await?,
181 output_root,
182 entrypoint,
183 }
184 .cell())
185}
186
187#[turbo_tasks::function(operation, root, session_dependent)]
188async fn create_evaluate_pool_assets_operation(
189 entries: ResolvedVc<EvaluateEntries>,
190 chunking_context: ResolvedVc<Box<dyn ChunkingContext>>,
191 module_graph: ResolvedVc<ModuleGraph>,
192) -> Result<Vc<EmittedEvaluatePoolAssets>> {
193 let operation = emit_evaluate_pool_assets_operation(entries, chunking_context, module_graph);
194 let assets = resolve_strongly_consistent_and_take_and_apply_effects(operation).await?;
204
205 Ok(*assets)
206}
207
208#[turbo_tasks::task_input]
209#[derive(Clone, Copy, Hash, Debug, PartialEq, Eq, TraceRawVcs, Encode, Decode)]
210pub enum EnvVarTracking {
211 WholeEnvTracked,
212 Untracked,
213}
214
215#[turbo_tasks::function(operation, root)]
216pub async fn get_evaluate_pool(
219 entries: ResolvedVc<EvaluateEntries>,
220 cwd: FileSystemPath,
221 env: ResolvedVc<Box<dyn ProcessEnv>>,
222 node_backend: ResolvedVc<Box<dyn NodeBackend>>,
223 chunking_context: ResolvedVc<Box<dyn ChunkingContext>>,
224 module_graph: ResolvedVc<ModuleGraph>,
225 additional_invalidation: ResolvedVc<Completion>,
226 debug: bool,
227 env_var_tracking: EnvVarTracking,
228) -> Result<Vc<EvaluatePool>> {
229 let assets_op = create_evaluate_pool_assets_operation(entries, chunking_context, module_graph);
230 let assets = assets_op.read_strongly_consistent().await?;
233
234 let EmittedEvaluatePoolAssets {
235 bootstrap,
236 output_root,
237 entrypoint,
238 } = &*assets;
239
240 let (Some(cwd), Some(entrypoint)) = (
241 to_sys_path(cwd.clone()).await?,
242 to_sys_path(entrypoint.clone()).await?,
243 ) else {
244 panic!("can only evaluate from a disk filesystem");
245 };
246
247 content_changed(Vc::upcast(**bootstrap)).await?;
249 let assets_for_source_mapping =
250 internal_assets_for_source_mapping(**bootstrap, output_root.clone())
251 .to_resolved()
252 .await?;
253 let env = match env_var_tracking {
254 EnvVarTracking::WholeEnvTracked => env.read_all().await?,
255 EnvVarTracking::Untracked => {
256 common_node_env(*env).await?;
258 for name in ["FORCE_COLOR", "NO_COLOR", "OPENSSL_CONF", "TZ"] {
259 env.read(name.into()).await?;
260 }
261
262 env.read_all().untracked().await?
263 }
264 };
265
266 let node_backend = node_backend.into_trait_ref().await?;
267 let pool = node_backend
268 .create_pool(CreatePoolOptions {
269 cwd,
270 entrypoint,
271 env: env.iter().map(|(k, v)| (k.clone(), v.clone())).collect(),
272 assets_for_source_mapping,
273 assets_root: output_root.clone(),
274 project_dir: chunking_context.root_path().owned().await?,
275 concurrency: available_parallelism().map_or(1, |v| v.get()),
276 debug,
277 })
278 .await?;
279 pool.pre_warm();
280 additional_invalidation.await?;
281 Ok(pool.cell())
282}
283
284#[turbo_tasks::function]
285async fn common_node_env(env: Vc<Box<dyn ProcessEnv>>) -> Result<Vc<EnvMap>> {
286 let mut filtered = FxIndexMap::default();
287 let env = env.read_all().await?;
288 for (key, value) in &*env {
289 let uppercase = key.to_uppercase();
290 for filter in &["NODE_", "UV_", "SSL_"] {
291 if uppercase.starts_with(filter) {
292 filtered.insert(key.clone(), value.clone());
293 break;
294 }
295 }
296 }
297 Ok(Vc::cell(filtered))
298}
299
300struct PoolErrorHandler;
301
302const MAX_FAST_ATTEMPTS: usize = 5;
304const MAX_ATTEMPTS: usize = MAX_FAST_ATTEMPTS * 2;
306
307impl futures_retry::ErrorHandler<anyhow::Error> for PoolErrorHandler {
308 type OutError = anyhow::Error;
309
310 fn handle(&mut self, attempt: usize, err: anyhow::Error) -> RetryPolicy<Self::OutError> {
311 if attempt >= MAX_ATTEMPTS {
312 RetryPolicy::ForwardError(err)
313 } else if attempt >= MAX_FAST_ATTEMPTS {
314 RetryPolicy::WaitRetry(Duration::from_secs(1))
315 } else {
316 RetryPolicy::Repeat
317 }
318 }
319}
320
321pub trait EvaluateContext {
322 type InfoMessage: DeserializeOwned;
323 type RequestMessage: DeserializeOwned;
324 type ResponseMessage: Serialize;
325 type State: Default;
326
327 fn pool(&self) -> OperationVc<EvaluatePool>;
328 fn keep_alive(&self) -> bool {
329 false
330 }
331 fn args(&self) -> &[ResolvedVc<JsonValue>];
332 fn cwd(&self) -> Vc<FileSystemPath>;
333 fn emit_error(
334 &self,
335 error: StructuredError,
336 pool: &EvaluatePool,
337 ) -> impl Future<Output = Result<()>> + Send;
338 fn info(
339 &self,
340 state: &mut Self::State,
341 data: Self::InfoMessage,
342 pool: &EvaluatePool,
343 ) -> impl Future<Output = Result<()>> + Send;
344 fn request(
345 &self,
346 state: &mut Self::State,
347 data: Self::RequestMessage,
348 pool: &EvaluatePool,
349 ) -> impl Future<Output = Result<Self::ResponseMessage>> + Send;
350 fn finish(
351 &self,
352 state: Self::State,
353 pool: &EvaluatePool,
354 ) -> impl Future<Output = Result<()>> + Send;
355
356 fn crash_context_prefix(&self) -> Option<RcStr> {
362 None
363 }
364}
365
366pub async fn custom_evaluate(evaluate_context: impl EvaluateContext) -> Result<Vc<Option<RcStr>>> {
367 let pool_op = evaluate_context.pool();
368 let mut state = Default::default();
369
370 let pool = pool_op.read_strongly_consistent().await?;
373
374 let args = evaluate_context.args().iter().try_join().await?;
375 let kill = !evaluate_context.keep_alive();
378
379 let (mut operation, _) = FutureRetry::new(
385 || async {
386 let mut operation = pool.operation().await?;
387 operation
388 .send(Bytes::from(serde_json::to_vec(
389 &EvalJavaScriptOutgoingMessage::Evaluate {
390 args: args.iter().map(|v| &**v).collect(),
391 },
392 )?))
393 .await?;
394 Ok(operation)
395 },
396 PoolErrorHandler,
397 )
398 .await
399 .map_err(|(e, _)| e)?;
400
401 let result = pull_operation(&mut operation, &pool, &evaluate_context, &mut state).await?;
405
406 evaluate_context.finish(state, &pool).await?;
407
408 if kill {
409 operation.wait_or_kill().await?;
410 }
411
412 Ok(Vc::cell(result.map(RcStr::from)))
413}
414
415#[turbo_tasks::value]
416pub struct EvaluateEntries {
417 entries: Vec<ResolvedVc<Box<dyn EvaluatableAsset + 'static>>>,
418 main_entry_ident: ResolvedVc<AssetIdent>,
419}
420
421#[turbo_tasks::value_impl]
422impl EvaluateEntries {
423 #[turbo_tasks::function]
424 pub async fn graph_entries(self: Vc<Self>) -> Result<Vc<GraphEntries>> {
425 Ok(
426 GraphEntries::from_chunk_groups(vec![ChunkGroupEntry::Entry {
427 modules: self
428 .await?
429 .entries
430 .iter()
431 .cloned()
432 .map(ResolvedVc::upcast)
433 .collect(),
434 heuristics: EntryHeuristics::default(),
435 }])
436 .cell(),
437 )
438 }
439}
440
441#[turbo_tasks::function]
442pub async fn get_evaluate_entries(
443 module_asset: ResolvedVc<Box<dyn Module>>,
444 asset_context: ResolvedVc<Box<dyn AssetContext>>,
445 node_backend: ResolvedVc<Box<dyn NodeBackend>>,
446 runtime_entries: Option<ResolvedVc<EvaluatableAssets>>,
447) -> Result<Vc<EvaluateEntries>> {
448 let node_backend = node_backend.into_trait_ref().await?;
449 let runtime_module_path = node_backend.runtime_module_path();
450
451 let runtime_asset = asset_context
452 .process(
453 Vc::upcast(FileSource::new(
454 embed_file_path(runtime_module_path).owned().await?,
455 )),
456 ReferenceType::Internal(InnerAssets::empty().to_resolved().await?),
457 )
458 .module()
459 .to_resolved()
460 .await?;
461
462 let entry_module = asset_context
463 .process(
464 Vc::upcast(VirtualSource::new(
465 runtime_asset.ident().await?.path.join("evaluate.js")?,
466 AssetContent::file(
467 FileContent::Content(File::from(
468 "import {run} from 'RUNTIME'; run(() => import('INNER'))",
469 ))
470 .cell(),
471 ),
472 )),
473 ReferenceType::Internal(ResolvedVc::cell(
474 fxindexmap! {rcstr!("INNER") => module_asset,
475 rcstr!("RUNTIME") => runtime_asset},
476 )),
477 )
478 .module()
479 .to_resolved()
480 .await?;
481
482 let runtime_entries = {
483 let mut entries = vec![];
484 let global_module_path = node_backend.globals_module_path();
485
486 let globals_module = asset_context
487 .process(
488 Vc::upcast(FileSource::new(
489 embed_file_path(global_module_path).owned().await?,
490 )),
491 ReferenceType::Internal(InnerAssets::empty().to_resolved().await?),
492 )
493 .module();
494
495 let Some(globals_module) = ResolvedVc::try_sidecast::<Box<dyn EvaluatableAsset>>(
496 globals_module.to_resolved().await?,
497 ) else {
498 bail!("Internal module is not evaluatable");
499 };
500
501 entries.push(globals_module);
502
503 if let Some(runtime_entries) = runtime_entries {
504 for &entry in &*runtime_entries.await? {
505 entries.push(entry)
506 }
507 }
508 entries
509 };
510
511 Ok(EvaluateEntries {
512 entries: runtime_entries
513 .iter()
514 .copied()
515 .chain(iter::once(ResolvedVc::try_downcast(entry_module).unwrap()))
516 .collect(),
517 main_entry_ident: module_asset.ident().to_resolved().await?,
518 }
519 .cell())
520}
521
522#[turbo_tasks::function]
525pub async fn evaluate(
526 entries: ResolvedVc<EvaluateEntries>,
527 cwd: FileSystemPath,
528 env: ResolvedVc<Box<dyn ProcessEnv>>,
529 node_backend: ResolvedVc<Box<dyn NodeBackend>>,
530 context_source_for_issue: ResolvedVc<Box<dyn Source>>,
531 chunking_context: ResolvedVc<Box<dyn ChunkingContext>>,
532 module_graph: ResolvedVc<ModuleGraph>,
533 args: Vec<ResolvedVc<JsonValue>>,
534 additional_invalidation: ResolvedVc<Completion>,
535 debug: bool,
536) -> Result<Vc<Option<RcStr>>> {
537 custom_evaluate(BasicEvaluateContext {
538 entries,
539 cwd,
540 env,
541 node_backend,
542 context_source_for_issue,
543 chunking_context,
544 module_graph,
545 args,
546 additional_invalidation,
547 debug,
548 })
549 .await
550}
551
552async fn pull_operation<T: EvaluateContext>(
555 operation: &mut Box<dyn Operation>,
556 pool: &EvaluatePool,
557 evaluate_context: &T,
558 state: &mut T::State,
559) -> Result<Option<String>> {
560 let _guard = duration_span!("Node.js evaluation");
561
562 loop {
563 let recv_result = operation.recv().await;
564 let bytes = match recv_result {
565 Ok(bytes) => bytes,
566 Err(err) => {
567 let message = match evaluate_context.crash_context_prefix() {
574 Some(prefix) => format!(
575 "Node.js subprocess crashed while evaluating {}: {}",
576 prefix,
577 PrettyPrintError(&err)
578 ),
579 None => format!(
580 "Node.js subprocess crashed while evaluating: {}",
581 PrettyPrintError(&err)
582 ),
583 };
584 let synthetic = StructuredError::from_message("Error".to_string(), message);
585 evaluate_context.emit_error(synthetic, pool).await?;
586 operation.disallow_reuse();
587 return Ok(None);
588 }
589 };
590 let message = serde_json::from_slice(&bytes)?;
591
592 match message {
593 EvalJavaScriptIncomingMessage::Error(error) => {
594 evaluate_context.emit_error(error, pool).await?;
595 operation.disallow_reuse();
597 return Ok(None);
599 }
600 EvalJavaScriptIncomingMessage::End { data } => return Ok(data),
601 EvalJavaScriptIncomingMessage::Info { data } => {
602 evaluate_context
603 .info(state, serde_json::from_value(data)?, pool)
604 .await?;
605 }
606 EvalJavaScriptIncomingMessage::Request { id, data } => {
607 match evaluate_context
608 .request(state, serde_json::from_value(data)?, pool)
609 .await
610 {
611 Ok(response) => {
612 operation
613 .send(Bytes::from(serde_json::to_vec(
614 &EvalJavaScriptOutgoingMessage::Result {
615 id,
616 error: None,
617 data: Some(serde_json::to_value(response)?),
618 },
619 )?))
620 .await?;
621 }
622 Err(e) => {
623 operation
624 .send(Bytes::from(serde_json::to_vec(
625 &EvalJavaScriptOutgoingMessage::Result {
626 id,
627 error: Some(PrettyPrintError(&e).to_string()),
628 data: None,
629 },
630 )?))
631 .await?;
632 }
633 }
634 }
635 }
636 }
637}
638
639struct BasicEvaluateContext {
640 entries: ResolvedVc<EvaluateEntries>,
641 cwd: FileSystemPath,
642 env: ResolvedVc<Box<dyn ProcessEnv>>,
643 node_backend: ResolvedVc<Box<dyn NodeBackend>>,
644 context_source_for_issue: ResolvedVc<Box<dyn Source>>,
645 chunking_context: ResolvedVc<Box<dyn ChunkingContext>>,
646 module_graph: ResolvedVc<ModuleGraph>,
647 args: Vec<ResolvedVc<JsonValue>>,
648 additional_invalidation: ResolvedVc<Completion>,
649 debug: bool,
650}
651
652impl EvaluateContext for BasicEvaluateContext {
653 type InfoMessage = ();
654 type RequestMessage = ();
655 type ResponseMessage = ();
656 type State = ();
657
658 fn pool(&self) -> OperationVc<EvaluatePool> {
659 get_evaluate_pool(
660 self.entries,
661 self.cwd.clone(),
662 self.env,
663 self.node_backend,
664 self.chunking_context,
665 self.module_graph,
666 self.additional_invalidation,
667 self.debug,
668 EnvVarTracking::WholeEnvTracked,
669 )
670 }
671
672 fn args(&self) -> &[ResolvedVc<serde_json::Value>] {
673 &self.args
674 }
675
676 fn cwd(&self) -> Vc<turbo_tasks_fs::FileSystemPath> {
677 self.cwd.clone().cell()
678 }
679
680 fn keep_alive(&self) -> bool {
681 !self.args.is_empty()
682 }
683
684 async fn emit_error(&self, error: StructuredError, pool: &EvaluatePool) -> Result<()> {
685 EvaluationIssue {
686 error,
687 source: IssueSource::from_source_only(self.context_source_for_issue),
688 assets_for_source_mapping: pool.assets_for_source_mapping,
689 assets_root: pool.assets_root.clone(),
690 root_path: self.chunking_context.root_path().owned().await?,
691 detail: None,
692 }
693 .resolved_cell()
694 .emit();
695 Ok(())
696 }
697
698 async fn info(
699 &self,
700 _state: &mut Self::State,
701 _data: Self::InfoMessage,
702 _pool: &EvaluatePool,
703 ) -> Result<()> {
704 bail!("BasicEvaluateContext does not support info messages")
705 }
706
707 async fn request(
708 &self,
709 _state: &mut Self::State,
710 _data: Self::RequestMessage,
711 _pool: &EvaluatePool,
712 ) -> Result<Self::ResponseMessage> {
713 bail!("BasicEvaluateContext does not support request messages")
714 }
715
716 async fn finish(&self, _state: Self::State, _pool: &EvaluatePool) -> Result<()> {
717 Ok(())
718 }
719}
720
721#[turbo_tasks::value(shared)]
723pub struct EvaluationIssue {
724 pub source: IssueSource,
725 pub error: StructuredError,
726 pub assets_for_source_mapping: ResolvedVc<AssetsForSourceMapping>,
727 pub assets_root: FileSystemPath,
728 pub root_path: FileSystemPath,
729 pub detail: Option<RcStr>,
733}
734
735#[async_trait]
736#[turbo_tasks::value_impl]
737impl Issue for EvaluationIssue {
738 async fn title(&self) -> Result<StyledString> {
739 Ok(StyledString::Text(rcstr!("Error evaluating Node.js code")))
740 }
741
742 fn stage(&self) -> IssueStage {
743 IssueStage::Transform
744 }
745
746 async fn file_path(&self) -> Result<FileSystemPath> {
747 self.source.file_path().await
748 }
749
750 async fn description(&self) -> Result<Option<StyledString>> {
751 Ok(Some(StyledString::Text(
752 self.error
753 .print(
754 *self.assets_for_source_mapping,
755 self.assets_root.clone(),
756 self.root_path.clone(),
757 FormattingMode::Plain,
758 )
759 .await?
760 .into(),
761 )))
762 }
763
764 async fn detail(&self) -> Result<Option<StyledString>> {
765 Ok(self.detail.clone().map(StyledString::Text))
766 }
767
768 fn source(&self) -> Option<IssueSource> {
769 Some(self.source)
770 }
771}