vercel · jridgewell · Mar 29, 2023 · Mar 21, 2023 · Mar 23, 2023 · Mar 24, 2023
@@ -52,6 +52,26 @@ impl BodyError {
     }
 }
 
+impl From<&str> for BodyError {
+    fn from(err: &str) -> Self {
+        BodyError {
+            err: err.to_string(),
+        }
+    }
+}
+
+impl From<String> for BodyError {
+    fn from(err: String) -> Self {
+        BodyError { err }
+    }
+}
+
+impl From<anyhow::Error> for BodyError {
+    fn from(value: anyhow::Error) -> Self {
+        value.to_string().into()
+    }
+}
+
 /// The return value of a content source when getting a path. A specificity is
 /// attached and when combining results this specificity should be used to order
 /// results.
@@ -246,29 +266,29 @@ pub struct ContentSourceData {
     pub cache_buster: u64,
 }
 
-type Chunk = Result<Bytes, BodyError>;
+pub type BodyChunk = Result<Bytes, BodyError>;
-pub type BodyChunk = Result<Bytes, BodyError>;
+pub type BodyChunk = Result<Bytes, SharedError>;
-pub type BodyChunk = Result<Bytes, BodyError>;
+pub type BodyChunk = Result<Bytes, SharedError>;
 /// A request body.
 #[turbo_tasks::value(shared)]
 #[derive(Default, Clone, Debug)]
 pub struct Body {
     #[turbo_tasks(trace_ignore)]
-    chunks: Stream<Chunk>,
+    chunks: Stream<BodyChunk>,
 }
 
 impl Body {
     /// Creates a new body from a list of chunks.
-    pub fn new(chunks: Vec<Chunk>) -> Self {
+    pub fn new(chunks: Vec<BodyChunk>) -> Self {
         Self {
             chunks: Stream::new_closed(chunks),
         }
     }
 
     /// Returns an iterator over the body's chunks.
-    pub fn read(&self) -> StreamRead<Chunk> {
+    pub fn read(&self) -> StreamRead<BodyChunk> {
         self.chunks.read()
     }
 
-    pub fn from_stream<T: StreamTrait<Item = Chunk> + Send + Sync + Unpin + 'static>(
+    pub fn from_stream<T: StreamTrait<Item = BodyChunk> + Send + Unpin + 'static>(
         source: T,
     ) -> Self {
         Self {

@@ -249,17 +249,6 @@ pub fn evaluate(
     additional_invalidation: CompletionVc,
     debug: bool,
 ) -> JavaScriptEvaluationVc {
-    let pool = get_evaluate_pool(
-        module_asset,
-        cwd,
-        env,
-        context,
-        chunking_context,
-        runtime_entries,
-        additional_invalidation,
-        debug,
-    );
-
     // Note the following code uses some hacks to create a child task that produces
     // a stream that is returned by this task.
 
@@ -278,11 +267,16 @@ pub fn evaluate(
 
     // run the evaluation as side effect
     compute_evaluate_stream(
-        pool,
+        module_asset,
         cwd,
+        env,
         context_ident_for_issue,
+        context,
         chunking_context,
+        runtime_entries,
         args,
+        additional_invalidation,
+        debug,
         JavaScriptStreamSender {
             get: Box::new(move || {
                 if let Some(sender) = initial.lock().take() {
@@ -308,11 +302,16 @@ pub fn evaluate(
 
 #[turbo_tasks::function]
 async fn compute_evaluate_stream(
-    pool: NodeJsPoolVc,
+    module_asset: AssetVc,
     cwd: FileSystemPathVc,
+    env: ProcessEnvVc,
     context_ident_for_issue: AssetIdentVc,
+    context: AssetContextVc,
     chunking_context: ChunkingContextVc,
+    runtime_entries: Option<EcmascriptChunkPlaceablesVc>,
     args: Vec<JsonValueVc>,
+    additional_invalidation: CompletionVc,
+    debug: bool,
     sender: JavaScriptStreamSenderVc,
 ) {
     mark_finished();
@@ -322,7 +321,20 @@ async fn compute_evaluate_stream(
     };
 
     let stream = generator! {
-        let pool = pool.await?;
+        let pool = get_evaluate_pool(
+            module_asset,
+            cwd,
+            env,
+            context,
+            chunking_context,
+            runtime_entries,
+            additional_invalidation,
+            debug,
+        );
+
+        // Read this strongly consistent, since we don't want to run inconsistent
+        // node.js code.
+        let pool = pool.strongly_consistent().await?;
 
         let args = args.into_iter().try_join().await?;
         // Assume this is a one-off operation, so we can kill the process
@@ -407,7 +419,7 @@ async fn pull_operation(
         match operation.recv().await? {
             EvalJavaScriptIncomingMessage::Error(error) => {
                 EvaluationIssue {
-                    error: error.clone(),
+                    error,
                     context_ident: context_ident_for_issue,
                     assets_for_source_mapping: pool.assets_for_source_mapping,
                     assets_root: pool.assets_root,

@@ -278,6 +278,7 @@ pub async fn get_intermediate_asset(
     .into())
 }
 
+#[derive(Clone, Debug)]
 #[turbo_tasks::value(shared)]
 pub struct ResponseHeaders {
     pub status: u16,

@@ -35,15 +35,16 @@ enum RenderProxyOutgoingMessage<'a> {
     BodyEnd,
 }
 
-#[derive(Deserialize)]
+#[derive(Deserialize, Debug)]
 #[serde(tag = "type", rename_all = "camelCase")]
 enum RenderProxyIncomingMessage {
     Headers { data: ResponseHeaders },
-    Body { data: Vec<u8> },
+    BodyChunk { data: Vec<u8> },
+    BodyEnd,
     Error(StructuredError),
 }
 
-#[derive(Deserialize)]
+#[derive(Deserialize, Debug)]
 #[serde(tag = "type", rename_all = "camelCase")]
 enum RenderStaticIncomingMessage {
     #[serde(rename_all = "camelCase")]
@@ -52,6 +53,13 @@ enum RenderStaticIncomingMessage {
         headers: Vec<(String, String)>,
         body: String,
     },
+    Headers {
+        data: ResponseHeaders,
+    },
+    BodyChunk {
+        data: Vec<u8>,
+    },
+    BodyEnd,
     Rewrite {
         path: String,
     },