Repository: denoland/deno
Pull request: fix(otel): don't panic when using worker threads

Closes https://github.com/denoland/deno/issues/28790

Changed files:
filename: runtime/ops/worker_host.rs
patch: @@ -19,6 +19,7 @@ use deno_web::MessagePortError;
 use log::debug;
 
 use crate::ops::TestingFeaturesEnabled;
+use crate::tokio_util::create_and_run_current_thread;
 use crate::web_worker::run_web_worker;
 use crate::web_worker::SendableWebWorkerHandle;
 use crate::web_worker::WebWorker;
@@ -206,33 +207,37 @@ fn op_create_worker(
     // - JS worker is useless - meaning it throws an exception and can't do anything else,
     //  all action done upon it should be noops
     // - newly spawned thread exits
+    let fut = async move {
+      let (worker, external_handle) =
+        (create_web_worker_cb.0)(CreateWebWorkerArgs {
+          name: worker_name,
+          worker_id,
+          parent_permissions,
+          permissions: worker_permissions,
+          main_module: module_specifier.clone(),
+          worker_type,
+          close_on_idle: args.close_on_idle,
+          maybe_worker_metadata,
+        });
+
+      // Send thread safe handle from newly created worker to host thread
+      handle_sender.send(external_handle).unwrap();
+      drop(handle_sender);
+
+      // At this point the only method of communication with host
+      // is using `worker.internal_channels`.
+      //
+      // Host can already push messages and interact with worker.
+      run_web_worker(
+        worker,
+        module_specifier,
+        maybe_source_code,
+        format_js_error_fn.0,
+      )
+      .await
+    };
 
-    let (worker, external_handle) =
-      (create_web_worker_cb.0)(CreateWebWorkerArgs {
-        name: worker_name,
-        worker_id,
-        parent_permissions,
-        permissions: worker_permissions,
-        main_module: module_specifier.clone(),
-        worker_type,
-        close_on_idle: args.close_on_idle,
-        maybe_worker_metadata,
-      });
-
-    // Send thread safe handle from newly created worker to host thread
-    handle_sender.send(external_handle).unwrap();
-    drop(handle_sender);
-
-    // At this point the only method of communication with host
-    // is using `worker.internal_channels`.
-    //
-    // Host can already push messages and interact with worker.
-    run_web_worker(
-      worker,
-      module_specifier,
-      maybe_source_code,
-      format_js_error_fn.0,
-    )
+    create_and_run_current_thread(fut)
   })?;
 
   // Receive WebWorkerHandle from newly created worker

filename: runtime/web_worker.rs
patch: @@ -61,7 +61,6 @@ use node_resolver::NpmPackageFolderResolver;
 use crate::inspector_server::InspectorServer;
 use crate::ops;
 use crate::shared::runtime;
-use crate::tokio_util::create_and_run_current_thread;
 use crate::worker::create_op_metrics;
 use crate::worker::import_meta_resolve_callback;
 use crate::worker::validate_import_attributes_callback;
@@ -957,66 +956,61 @@ fn print_worker_error(
 
 /// This function should be called from a thread dedicated to this worker.
 // TODO(bartlomieju): check if order of actions is aligned to Worker spec
-pub fn run_web_worker(
+// TODO(bartlomieju): run following block using "select!"
+// with terminate
+pub async fn run_web_worker(
   mut worker: WebWorker,
   specifier: ModuleSpecifier,
   mut maybe_source_code: Option<String>,
   format_js_error_fn: Option<Arc<FormatJsErrorFn>>,
 ) -> Result<(), CoreError> {
   let name = worker.name.to_string();
-
-  // TODO(bartlomieju): run following block using "select!"
-  // with terminate
-
-  let fut = async move {
-    let internal_handle = worker.internal_handle.clone();
-
-    // Execute provided source code immediately
-    let result = if let Some(source_code) = maybe_source_code.take() {
-      let r = worker.execute_script(located_script_name!(), source_code.into());
-      worker.start_polling_for_messages();
-      r
-    } else {
-      // TODO(bartlomieju): add "type": "classic", ie. ability to load
-      // script instead of module
-      match worker.preload_main_module(&specifier).await {
-        Ok(id) => {
-          worker.start_polling_for_messages();
-          worker.execute_main_module(id).await
-        }
-        Err(e) => Err(e),
+  let internal_handle = worker.internal_handle.clone();
+
+  // Execute provided source code immediately
+  let result = if let Some(source_code) = maybe_source_code.take() {
+    let r = worker.execute_script(located_script_name!(), source_code.into());
+    worker.start_polling_for_messages();
+    r
+  } else {
+    // TODO(bartlomieju): add "type": "classic", ie. ability to load
+    // script instead of module
+    match worker.preload_main_module(&specifier).await {
+      Ok(id) => {
+        worker.start_polling_for_messages();
+        worker.execute_main_module(id).await
       }
-    };
-
-    // If sender is closed it means that worker has already been closed from
-    // within using "globalThis.close()"
-    if internal_handle.is_terminated() {
-      return Ok(());
+      Err(e) => Err(e),
     }
+  };
 
-    let result = if result.is_ok() {
-      worker
-        .run_event_loop(PollEventLoopOptions {
-          wait_for_inspector: true,
-          ..Default::default()
-        })
-        .await
-    } else {
-      result
-    };
-
-    if let Err(e) = result {
-      print_worker_error(&e, &name, format_js_error_fn.as_deref());
-      internal_handle
-        .post_event(WorkerControlEvent::TerminalError(e))
-        .expect("Failed to post message to host");
-
-      // Failure to execute script is a terminal error, bye, bye.
-      return Ok(());
-    }
+  // If sender is closed it means that worker has already been closed from
+  // within using "globalThis.close()"
+  if internal_handle.is_terminated() {
+    return Ok(());
+  }
 
-    debug!("Worker thread shuts down {}", &name);
+  let result = if result.is_ok() {
+    worker
+      .run_event_loop(PollEventLoopOptions {
+        wait_for_inspector: true,
+        ..Default::default()
+      })
+      .await
+  } else {
     result
   };
-  create_and_run_current_thread(fut)
+
+  if let Err(e) = result {
+    print_worker_error(&e, &name, format_js_error_fn.as_deref());
+    internal_handle
+      .post_event(WorkerControlEvent::TerminalError(e))
+      .expect("Failed to post message to host");
+
+    // Failure to execute script is a terminal error, bye, bye.
+    return Ok(());
+  }
+
+  debug!("Worker thread shuts down {}", &name);
+  result
 }