fix: op_set_exit_code (#13034)

Fixes "op_set_exit_code" by sharing a single "Arc" between
all workers (via "op state") instead of having a "global" value stored in
"deno_runtime" crate. As a consequence setting an exit code is always
scoped to a tree of workers, instead of being overridable if there are
multiple worker tree (like in "deno test --jobs" subcommand).

Refactored "cli/main.rs" functions to return "Result<i32, AnyError>" instead
of "Result<(), AnyError>" so they can return exit code.

10 files changed, 131 insertions, 70 deletions

diff --git a/cli/main.rs b/cli/main.rs
index 8878fb57a..891aa1ac5 100644
--- a/cli/main.rs
+++ b/cli/main.rs
@@ -94,7 +94,6 @@ use std::iter::once;
 use std::path::PathBuf;
 use std::pin::Pin;
 use std::rc::Rc;
-use std::sync::atomic::Ordering::Relaxed;
 use std::sync::Arc;
 
 fn create_web_worker_callback(ps: ProcState) -> Arc<CreateWebWorkerCb> {
@@ -149,6 +148,7 @@ fn create_web_worker_callback(ps: ProcState) -> Arc<CreateWebWorkerCb> {
       broadcast_channel: ps.broadcast_channel.clone(),
       shared_array_buffer_store: Some(ps.shared_array_buffer_store.clone()),
       compiled_wasm_module_store: Some(ps.compiled_wasm_module_store.clone()),
+      maybe_exit_code: args.maybe_exit_code,
     };
     let bootstrap_options = options.bootstrap.clone();
 
@@ -406,7 +406,7 @@ pub fn get_types(unstable: bool) -> String {
 async fn compile_command(
   flags: Flags,
   compile_flags: CompileFlags,
-) -> Result<(), AnyError> {
+) -> Result<i32, AnyError> {
   let debug = flags.log_level == Some(log::Level::Debug);
 
   let run_flags = tools::standalone::compile_to_runtime_flags(
@@ -466,13 +466,13 @@ async fn compile_command(
   )
   .await?;
 
-  Ok(())
+  Ok(0)
 }
 
 async fn info_command(
   flags: Flags,
   info_flags: InfoFlags,
-) -> Result<(), AnyError> {
+) -> Result<i32, AnyError> {
   let ps = ProcState::build(flags).await?;
   if let Some(specifier) = info_flags.file {
     let specifier = resolve_url_or_path(&specifier)?;
@@ -512,21 +512,21 @@ async fn info_command(
     .await;
 
     if info_flags.json {
-      write_json_to_stdout(&json!(graph))
+      write_json_to_stdout(&json!(graph))?;
     } else {
-      write_to_stdout_ignore_sigpipe(graph.to_string().as_bytes())
-        .map_err(|err| err.into())
+      write_to_stdout_ignore_sigpipe(graph.to_string().as_bytes())?;
     }
   } else {
     // If it was just "deno info" print location of caches and exit
-    print_cache_info(&ps, info_flags.json, ps.flags.location.as_ref())
+    print_cache_info(&ps, info_flags.json, ps.flags.location.as_ref())?;
   }
+  Ok(0)
 }
 
 async fn install_command(
   flags: Flags,
   install_flags: InstallFlags,
-) -> Result<(), AnyError> {
+) -> Result<i32, AnyError> {
   let mut preload_flags = flags.clone();
   preload_flags.inspect = None;
   preload_flags.inspect_brk = None;
@@ -544,27 +544,30 @@ async fn install_command(
     install_flags.name,
     install_flags.root,
     install_flags.force,
-  )
+  )?;
+  Ok(0)
 }
 
 async fn uninstall_command(
   uninstall_flags: UninstallFlags,
-) -> Result<(), AnyError> {
-  tools::installer::uninstall(uninstall_flags.name, uninstall_flags.root)
+) -> Result<i32, AnyError> {
+  tools::installer::uninstall(uninstall_flags.name, uninstall_flags.root)?;
+  Ok(0)
 }
 
-async fn lsp_command() -> Result<(), AnyError> {
-  lsp::start().await
+async fn lsp_command() -> Result<i32, AnyError> {
+  lsp::start().await?;
+  Ok(0)
 }
 
 #[allow(clippy::too_many_arguments)]
 async fn lint_command(
   flags: Flags,
   lint_flags: LintFlags,
-) -> Result<(), AnyError> {
+) -> Result<i32, AnyError> {
   if lint_flags.rules {
     tools::lint::print_rules_list(lint_flags.json);
-    return Ok(());
+    return Ok(0);
   }
 
   let ps = ProcState::build(flags.clone()).await?;
@@ -574,13 +577,14 @@ async fn lint_command(
     None
   };
 
-  tools::lint::lint(maybe_lint_config, lint_flags, flags.watch).await
+  tools::lint::lint(maybe_lint_config, lint_flags, flags.watch).await?;
+  Ok(0)
 }
 
 async fn cache_command(
   flags: Flags,
   cache_flags: CacheFlags,
-) -> Result<(), AnyError> {
+) -> Result<i32, AnyError> {
   let lib = if flags.unstable {
     emit::TypeLib::UnstableDenoWindow
   } else {
@@ -601,13 +605,13 @@ async fn cache_command(
     .await?;
   }
 
-  Ok(())
+  Ok(0)
 }
 
 async fn eval_command(
   flags: Flags,
   eval_flags: EvalFlags,
-) -> Result<(), AnyError> {
+) -> Result<i32, AnyError> {
   // deno_graph works off of extensions for local files to determine the media
   // type, and so our "fake" specifier needs to have the proper extension.
   let main_module =
@@ -650,7 +654,7 @@ async fn eval_command(
     &located_script_name!(),
     "window.dispatchEvent(new Event('unload'))",
   )?;
-  Ok(())
+  Ok(0)
 }
 
 async fn create_graph_and_maybe_check(
@@ -801,7 +805,7 @@ fn human_size(size: f64) -> String {
 async fn bundle_command(
   flags: Flags,
   bundle_flags: BundleFlags,
-) -> Result<(), AnyError> {
+) -> Result<i32, AnyError> {
   let debug = flags.log_level == Some(log::Level::Debug);
 
   let resolver = |_| {
@@ -902,13 +906,13 @@ async fn bundle_command(
     operation(module_graph).await?;
   }
 
-  Ok(())
+  Ok(0)
 }
 
 async fn doc_command(
   flags: Flags,
   doc_flags: DocFlags,
-) -> Result<(), AnyError> {
+) -> Result<i32, AnyError> {
   tools::doc::print_docs(
     flags,
     doc_flags.source_file,
@@ -916,13 +920,14 @@ async fn doc_command(
     doc_flags.filter,
     doc_flags.private,
   )
-  .await
+  .await?;
+  Ok(0)
 }
 
 async fn format_command(
   flags: Flags,
   fmt_flags: FmtFlags,
-) -> Result<(), AnyError> {
+) -> Result<i32, AnyError> {
   let ps = ProcState::build(flags.clone()).await?;
   let maybe_fmt_config = if let Some(config_file) = &ps.maybe_config_file {
     config_file.to_fmt_config()?
@@ -931,17 +936,21 @@ async fn format_command(
   };
 
   if fmt_flags.files.len() == 1 && fmt_flags.files[0].to_string_lossy() == "-" {
-    return tools::fmt::format_stdin(
+    tools::fmt::format_stdin(
       fmt_flags,
       maybe_fmt_config.map(|c| c.options).unwrap_or_default(),
-    );
+    )?;
+    return Ok(0);
   }
 
   tools::fmt::format(fmt_flags, flags.watch, maybe_fmt_config).await?;
-  Ok(())
+  Ok(0)
 }
 
-async fn run_repl(flags: Flags, repl_flags: ReplFlags) -> Result<(), AnyError> {
+async fn repl_command(
+  flags: Flags,
+  repl_flags: ReplFlags,
+) -> Result<i32, AnyError> {
   let main_module = resolve_url_or_path("./$deno$repl.ts").unwrap();
   let permissions = Permissions::from_options(&flags.clone().into());
   let ps = ProcState::build(flags.clone()).await?;
@@ -956,7 +965,7 @@ async fn run_repl(flags: Flags, repl_flags: ReplFlags) -> Result<(), AnyError> {
   tools::repl::run(&ps, worker, repl_flags.eval).await
 }
 
-async fn run_from_stdin(flags: Flags) -> Result<(), AnyError> {
+async fn run_from_stdin(flags: Flags) -> Result<i32, AnyError> {
   let ps = ProcState::build(flags.clone()).await?;
   let permissions = Permissions::from_options(&flags.clone().into());
   let main_module = resolve_url_or_path("./$deno$stdin.ts").unwrap();
@@ -992,10 +1001,12 @@ async fn run_from_stdin(flags: Flags) -> Result<(), AnyError> {
     &located_script_name!(),
     "window.dispatchEvent(new Event('unload'))",
   )?;
-  Ok(())
+  Ok(worker.get_exit_code())
 }
 
-async fn run_with_watch(flags: Flags, script: String) -> Result<(), AnyError> {
+// TODO(bartlomieju): this function is not handling `exit_code` set by the runtime
+// code properly.
+async fn run_with_watch(flags: Flags, script: String) -> Result<i32, AnyError> {
   let resolver = |_| {
     let script1 = script.clone();
     let script2 = script.clone();
@@ -1156,13 +1167,14 @@ async fn run_with_watch(flags: Flags, script: String) -> Result<(), AnyError> {
     }
   };
 
-  file_watcher::watch_func(resolver, operation, "Process").await
+  file_watcher::watch_func(resolver, operation, "Process").await?;
+  Ok(0)
 }
 
 async fn run_command(
   flags: Flags,
   run_flags: RunFlags,
-) -> Result<(), AnyError> {
+) -> Result<i32, AnyError> {
   // Read script content from stdin
   if run_flags.script == "-" {
     return run_from_stdin(flags).await;
@@ -1247,13 +1259,13 @@ async fn run_command(
       .with_event_loop(coverage_collector.stop_collecting().boxed_local())
       .await?;
   }
-  Ok(())
+  Ok(worker.get_exit_code())
 }
 
 async fn coverage_command(
   flags: Flags,
   coverage_flags: CoverageFlags,
-) -> Result<(), AnyError> {
+) -> Result<i32, AnyError> {
   if coverage_flags.files.is_empty() {
     return Err(generic_error("No matching coverage profiles found"));
   }
@@ -1266,13 +1278,14 @@ async fn coverage_command(
     coverage_flags.exclude,
     coverage_flags.lcov,
   )
-  .await
+  .await?;
+  Ok(0)
 }
 
 async fn test_command(
   flags: Flags,
   test_flags: TestFlags,
-) -> Result<(), AnyError> {
+) -> Result<i32, AnyError> {
   if let Some(ref coverage_dir) = flags.coverage_dir {
     std::fs::create_dir_all(&coverage_dir)?;
     env::set_var(
@@ -1295,7 +1308,7 @@ async fn test_command(
     )
     .await?;
 
-    return Ok(());
+    return Ok(0);
   }
 
   tools::test::run_tests(
@@ -1312,7 +1325,7 @@ async fn test_command(
   )
   .await?;
 
-  Ok(())
+  Ok(0)
 }
 
 fn init_v8_flags(v8_flags: &[String]) {
@@ -1341,7 +1354,7 @@ fn init_v8_flags(v8_flags: &[String]) {
 
 fn get_subcommand(
   flags: Flags,
-) -> Pin<Box<dyn Future<Output = Result<(), AnyError>>>> {
+) -> Pin<Box<dyn Future<Output = Result<i32, AnyError>>>> {
   match flags.clone().subcommand {
     DenoSubcommand::Bundle(bundle_flags) => {
       bundle_command(flags, bundle_flags).boxed_local()
@@ -1378,7 +1391,7 @@ fn get_subcommand(
       lint_command(flags, lint_flags).boxed_local()
     }
     DenoSubcommand::Repl(repl_flags) => {
-      run_repl(flags, repl_flags).boxed_local()
+      repl_command(flags, repl_flags).boxed_local()
     }
     DenoSubcommand::Run(run_flags) => {
       run_command(flags, run_flags).boxed_local()
@@ -1410,9 +1423,13 @@ fn get_subcommand(
         output,
         ca_file,
       } = upgrade_flags;
-      tools::upgrade::upgrade_command(
-        dry_run, force, canary, version, output, ca_file,
-      )
+      async move {
+        tools::upgrade::upgrade_command(
+          dry_run, force, canary, version, output, ca_file,
+        )
+        .await?;
+        Ok(0)
+      }
       .boxed_local()
     }
   }
@@ -1477,8 +1494,7 @@ pub fn main() {
 
   logger::init(flags.log_level);
 
-  unwrap_or_exit(run_basic(get_subcommand(flags)));
+  let exit_code = unwrap_or_exit(run_basic(get_subcommand(flags)));
 
-  let code = deno_runtime::EXIT_CODE.load(Relaxed);
-  std::process::exit(code);
+  std::process::exit(exit_code);
 }
diff --git a/cli/tests/integration/run_tests.rs b/cli/tests/integration/run_tests.rs
index fbdc01ad7..dd4482423 100644
--- a/cli/tests/integration/run_tests.rs
+++ b/cli/tests/integration/run_tests.rs
@@ -907,6 +907,12 @@ itest!(set_exit_code_2 {
   exit_code: 42,
 });
 
+itest!(set_exit_code_in_worker {
+  args: "run --no-check --unstable --allow-read set_exit_code_in_worker.ts",
+  output: "empty.out",
+  exit_code: 42,
+});
+
 itest!(heapstats {
   args: "run --quiet --unstable --v8-flags=--expose-gc heapstats.js",
   output: "heapstats.js.out",
diff --git a/cli/tests/testdata/set_exit_code_in_worker.ts b/cli/tests/testdata/set_exit_code_in_worker.ts
new file mode 100644
index 000000000..1df6a76d0
--- /dev/null
+++ b/cli/tests/testdata/set_exit_code_in_worker.ts
@@ -0,0 +1,13 @@
+import { deferred } from "../../../test_util/std/async/deferred.ts";
+
+const worker = new Worker(
+  new URL("set_exit_code_worker.js", import.meta.url).href,
+  { type: "module", deno: { namespace: true } },
+);
+
+const promise1 = deferred();
+worker.onmessage = (_e) => {
+  promise1.resolve();
+};
+await promise1;
+worker.terminate();
diff --git a/cli/tests/testdata/set_exit_code_worker.js b/cli/tests/testdata/set_exit_code_worker.js
new file mode 100644
index 000000000..d1c0123d1
--- /dev/null
+++ b/cli/tests/testdata/set_exit_code_worker.js
@@ -0,0 +1,4 @@
+// Set exit code
+Deno.core.opSync("op_set_exit_code", 42);
+
+self.postMessage("ok");
diff --git a/cli/tools/repl/mod.rs b/cli/tools/repl/mod.rs
index 047b477cf..687e8a300 100644
--- a/cli/tools/repl/mod.rs
+++ b/cli/tools/repl/mod.rs
@@ -747,7 +747,7 @@ pub async fn run(
   ps: &ProcState,
   worker: MainWorker,
   maybe_eval: Option<String>,
-) -> Result<(), AnyError> {
+) -> Result<i32, AnyError> {
   let mut repl_session = ReplSession::initialize(worker).await?;
   let mut rustyline_channel = rustyline_channel();
 
@@ -807,5 +807,5 @@ pub async fn run(
 
   editor.save_history()?;
 
-  Ok(())
+  Ok(repl_session.worker.get_exit_code())
 }
diff --git a/runtime/lib.rs b/runtime/lib.rs
index ad4a790da..58de3725b 100644
--- a/runtime/lib.rs
+++ b/runtime/lib.rs
@@ -1,7 +1,5 @@
 // Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.
 
-use std::sync::atomic::AtomicI32;
-
 pub use deno_broadcast_channel;
 pub use deno_console;
 pub use deno_core;
@@ -32,12 +30,3 @@ pub mod worker;
 
 mod worker_bootstrap;
 pub use worker_bootstrap::BootstrapOptions;
-
-// The global may not be very elegant but:
-//
-// 1. op_exit() calls std::process::exit() so there is not much point storing
-//    the exit code in runtime state
-//
-// 2. storing it in runtime state makes retrieving it again in cli/main.rs
-//    unduly complicated
-pub static EXIT_CODE: AtomicI32 = AtomicI32::new(0);
diff --git a/runtime/ops/os.rs b/runtime/ops/os.rs
index 8e96de75d..81f8a529e 100644
--- a/runtime/ops/os.rs
+++ b/runtime/ops/os.rs
@@ -10,9 +10,11 @@ use deno_core::OpState;
 use serde::Serialize;
 use std::collections::HashMap;
 use std::env;
+use std::sync::atomic::AtomicI32;
 use std::sync::atomic::Ordering::Relaxed;
+use std::sync::Arc;
 
-pub fn init() -> Extension {
+pub fn init(maybe_exit_code: Option<Arc<AtomicI32>>) -> Extension {
   Extension::builder()
     .ops(vec![
       ("op_exit", op_sync(op_exit)),
@@ -27,6 +29,11 @@ pub fn init() -> Extension {
       ("op_set_exit_code", op_sync(op_set_exit_code)),
       ("op_system_memory_info", op_sync(op_system_memory_info)),
     ])
+    .state(move |state| {
+      let exit_code = maybe_exit_code.clone().unwrap_or_default();
+      state.put::<Arc<AtomicI32>>(exit_code);
+      Ok(())
+    })
     .build()
 }
 
@@ -97,13 +104,17 @@ fn op_delete_env(
   Ok(())
 }
 
-fn op_set_exit_code(_: &mut OpState, code: i32, _: ()) -> Result<(), AnyError> {
-  crate::EXIT_CODE.store(code, Relaxed);
+fn op_set_exit_code(
+  state: &mut OpState,
+  code: i32,
+  _: (),
+) -> Result<(), AnyError> {
+  state.borrow_mut::<Arc<AtomicI32>>().store(code, Relaxed);
   Ok(())
 }
 
-fn op_exit(_: &mut OpState, _: (), _: ()) -> Result<(), AnyError> {
-  let code = crate::EXIT_CODE.load(Relaxed);
+fn op_exit(state: &mut OpState, _: (), _: ()) -> Result<(), AnyError> {
+  let code = state.borrow::<Arc<AtomicI32>>().load(Relaxed);
   std::process::exit(code)
 }
 
diff --git a/runtime/ops/worker_host.rs b/runtime/ops/worker_host.rs
index e9eb380e0..f290b3833 100644
--- a/runtime/ops/worker_host.rs
+++ b/runtime/ops/worker_host.rs
@@ -23,6 +23,7 @@ use log::debug;
 use std::cell::RefCell;
 use std::collections::HashMap;
 use std::rc::Rc;
+use std::sync::atomic::AtomicI32;
 use std::sync::Arc;
 use std::thread::JoinHandle;
 
@@ -34,6 +35,7 @@ pub struct CreateWebWorkerArgs {
   pub main_module: ModuleSpecifier,
   pub use_deno_namespace: bool,
   pub worker_type: WebWorkerType,
+  pub maybe_exit_code: Option<Arc<AtomicI32>>,
 }
 
 pub type CreateWebWorkerCb = dyn Fn(CreateWebWorkerArgs) -> (WebWorker, SendableWebWorkerHandle)
@@ -166,7 +168,11 @@ fn op_create_worker(
     parent_permissions.clone()
   };
   let parent_permissions = parent_permissions.clone();
-
+  // `try_borrow` here, because worker might have been started without
+  // access to `Deno` namespace.
+  // TODO(bartlomieju): can a situation happen when parent doesn't
+  // have access to `exit_code` but the child does?
+  let maybe_exit_code = state.try_borrow::<Arc<AtomicI32>>().cloned();
   let worker_id = state.take::<WorkerId>();
   let create_module_loader = state.take::<CreateWebWorkerCbHolder>();
   state.put::<CreateWebWorkerCbHolder>(create_module_loader.clone());
@@ -199,6 +205,7 @@ fn op_create_worker(
         main_module: module_specifier.clone(),
         use_deno_namespace,
         worker_type,
+        maybe_exit_code,
       });
 
     // Send thread safe handle from newly created worker to host thread
diff --git a/runtime/web_worker.rs b/runtime/web_worker.rs
index f5df2f58b..24306fab5 100644
--- a/runtime/web_worker.rs
+++ b/runtime/web_worker.rs
@@ -38,6 +38,7 @@ use std::cell::RefCell;
 use std::fmt;
 use std::rc::Rc;
 use std::sync::atomic::AtomicBool;
+use std::sync::atomic::AtomicI32;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::task::Context;
@@ -323,6 +324,7 @@ pub struct WebWorkerOptions {
   pub broadcast_channel: InMemoryBroadcastChannel,
   pub shared_array_buffer_store: Option<SharedArrayBufferStore>,
   pub compiled_wasm_module_store: Option<CompiledWasmModuleStore>,
+  pub maybe_exit_code: Option<Arc<AtomicI32>>,
 }
 
 impl WebWorker {
@@ -408,7 +410,9 @@ impl WebWorker {
           unstable,
           options.unsafely_ignore_certificate_errors.clone(),
         ),
-        ops::os::init(),
+        ops::os::init(Some(options.maybe_exit_code.expect(
+          "Worker has access to OS ops but exit code was not passed.",
+        ))),
         ops::permissions::init(),
         ops::process::init(),
         ops::signal::init(),
diff --git a/runtime/worker.rs b/runtime/worker.rs
index 3a16b7b55..545128dff 100644
--- a/runtime/worker.rs
+++ b/runtime/worker.rs
@@ -25,6 +25,8 @@ use deno_web::BlobStore;
 use log::debug;
 use std::pin::Pin;
 use std::rc::Rc;
+use std::sync::atomic::AtomicI32;
+use std::sync::atomic::Ordering::Relaxed;
 use std::sync::Arc;
 use std::task::Context;
 use std::task::Poll;
@@ -135,7 +137,7 @@ impl MainWorker {
         unstable,
         options.unsafely_ignore_certificate_errors.clone(),
       ),
-      ops::os::init(),
+      ops::os::init(None),
       ops::permissions::init(),
       ops::process::init(),
       ops::signal::init(),
@@ -289,6 +291,15 @@ impl MainWorker {
       };
     }
   }
+
+  /// Return exit code set by the executed code (either in main worker
+  /// or one of child web workers).
+  pub fn get_exit_code(&mut self) -> i32 {
+    let op_state_rc = self.js_runtime.op_state();
+    let op_state = op_state_rc.borrow();
+    let exit_code = op_state.borrow::<Arc<AtomicI32>>().load(Relaxed);
+    exit_code
+  }
 }
 
 #[cfg(test)]