Repository: denoland/deno
Pull request: fix(lint): out of order diagnostics for plugins

This commit fixes racy condition in lint plugins
that could have caused diagnostics for another
file to be printed for completely unrelated file.

With this change, a oneshot channel is used
to receive diagnostics for a file, which ensures
that the caller will receive diagnostics for itself.

Changed files:
filename: cli/tools/lint/plugins.rs
patch: @@ -25,9 +25,8 @@ use deno_runtime::deno_permissions::PermissionsContainer;
 use deno_runtime::tokio_util;
 use deno_runtime::worker::MainWorker;
 use deno_runtime::WorkerExecutionMode;
-use tokio::sync::mpsc::channel;
-use tokio::sync::mpsc::Receiver;
-use tokio::sync::mpsc::Sender;
+use tokio::sync::mpsc;
+use tokio::sync::oneshot;
 
 use crate::args::DenoSubcommand;
 use crate::args::Flags;
@@ -42,13 +41,15 @@ pub enum PluginHostRequest {
   LoadPlugins {
     specifiers: Vec<ModuleSpecifier>,
     exclude_rules: Option<Vec<String>>,
+    tx: oneshot::Sender<PluginHostResponse>,
   },
   Run {
     serialized_ast: Vec<u8>,
     file_path: PathBuf,
     source_text_info: SourceTextInfo,
     utf16_map: Utf16Map,
     maybe_token: Option<CancellationToken>,
+    tx: oneshot::Sender<PluginHostResponse>,
   },
 }
 
@@ -102,8 +103,7 @@ v8_static_strings! {
 
 #[derive(Debug)]
 pub struct PluginHostProxy {
-  tx: Sender<PluginHostRequest>,
-  rx: Arc<tokio::sync::Mutex<Receiver<PluginHostResponse>>>,
+  tx: mpsc::Sender<PluginHostRequest>,
   pub(crate) plugin_info: Arc<Mutex<Vec<PluginInfo>>>,
   #[allow(unused)]
   join_handle: std::thread::JoinHandle<Result<(), AnyError>>,
@@ -127,14 +127,12 @@ pub struct PluginHost {
   worker: MainWorker,
   install_plugins_fn: Rc<v8::Global<v8::Function>>,
   run_plugins_for_file_fn: Rc<v8::Global<v8::Function>>,
-  tx: Sender<PluginHostResponse>,
-  rx: Receiver<PluginHostRequest>,
+  rx: mpsc::Receiver<PluginHostRequest>,
 }
 
 async fn create_plugin_runner_inner(
   logger: PluginLogger,
-  rx_req: Receiver<PluginHostRequest>,
-  tx_res: Sender<PluginHostResponse>,
+  rx_req: mpsc::Receiver<PluginHostRequest>,
 ) -> Result<PluginHost, AnyError> {
   let flags = Flags {
     subcommand: DenoSubcommand::Lint(LintFlags::default()),
@@ -202,7 +200,6 @@ async fn create_plugin_runner_inner(
     worker,
     install_plugins_fn,
     run_plugins_for_file_fn,
-    tx: tx_res,
     rx: rx_req,
   })
 }
@@ -228,17 +225,15 @@ impl PluginInfo {
 
 impl PluginHost {
   fn create(logger: PluginLogger) -> Result<PluginHostProxy, AnyError> {
-    let (tx_req, rx_req) = channel(10);
-    let (tx_res, rx_res) = channel(10);
+    let (tx_req, rx_req) = mpsc::channel(10);
 
     let logger_ = logger.clone();
     let join_handle = std::thread::spawn(move || {
       let logger = logger_;
       log::debug!("Lint PluginHost thread spawned");
       let start = std::time::Instant::now();
       let fut = async move {
-        let runner =
-          create_plugin_runner_inner(logger.clone(), rx_req, tx_res).await?;
+        let runner = create_plugin_runner_inner(logger.clone(), rx_req).await?;
         log::debug!("Lint PlugibnHost running loop");
         runner.run_loop().await?;
         log::debug!(
@@ -253,7 +248,6 @@ impl PluginHost {
 
     let proxy = PluginHostProxy {
       tx: tx_req,
-      rx: Arc::new(tokio::sync::Mutex::new(rx_res)),
       plugin_info: Arc::new(Mutex::new(vec![])),
       join_handle,
     };
@@ -269,16 +263,18 @@ impl PluginHost {
         PluginHostRequest::LoadPlugins {
           specifiers,
           exclude_rules,
+          tx,
         } => {
           let r = self.load_plugins(specifiers, exclude_rules).await;
-          let _ = self.tx.send(PluginHostResponse::LoadPlugin(r)).await;
+          let _ = tx.send(PluginHostResponse::LoadPlugin(r));
         }
         PluginHostRequest::Run {
           serialized_ast,
           file_path,
           source_text_info,
           utf16_map,
           maybe_token,
+          tx,
         } => {
           let start = std::time::Instant::now();
           let r = match self.run_plugins(
@@ -295,7 +291,7 @@ impl PluginHost {
             "Running plugins lint rules took {:?}",
             std::time::Instant::now() - start
           );
-          let _ = self.tx.send(PluginHostResponse::Run(r)).await;
+          let _ = tx.send(PluginHostResponse::Run(r));
         }
       }
     }
@@ -452,16 +448,17 @@ impl PluginHostProxy {
     specifiers: Vec<ModuleSpecifier>,
     exclude_rules: Option<Vec<String>>,
   ) -> Result<(), AnyError> {
+    let (tx, rx) = oneshot::channel();
     self
       .tx
       .send(PluginHostRequest::LoadPlugins {
         specifiers,
         exclude_rules,
+        tx,
       })
       .await?;
-    let mut rx = self.rx.lock().await;
 
-    if let Some(val) = rx.recv().await {
+    if let Ok(val) = rx.await {
       let PluginHostResponse::LoadPlugin(result) = val else {
         unreachable!()
       };
@@ -480,6 +477,7 @@ impl PluginHostProxy {
     utf16_map: Utf16Map,
     maybe_token: Option<CancellationToken>,
   ) -> Result<Vec<LintDiagnostic>, AnyError> {
+    let (tx, rx) = oneshot::channel();
     self
       .tx
       .send(PluginHostRequest::Run {
@@ -488,11 +486,11 @@ impl PluginHostProxy {
         source_text_info,
         utf16_map,
         maybe_token,
+        tx,
       })
       .await?;
-    let mut rx = self.rx.lock().await;
 
-    if let Some(PluginHostResponse::Run(diagnostics_result)) = rx.recv().await {
+    if let Ok(PluginHostResponse::Run(diagnostics_result)) = rx.await {
       return diagnostics_result;
     }
     bail!("Plugin host has closed")