refactor(cli): move op sanitizer to Rust (#22413)

The format of the sanitizers will change a little bit:

- If multiple async ops leak and traces are on, we repeat the async op
header once per stack trace.
- All leaks are aggregated under a "Leaks detected:" banner as the new
timers are eventually going to be added, and these are neither ops nor
resources.
 - `1 async op` is now `An async op`
- If ops and resources leak, we show both (rather than op leaks masking
resources)

Follow-on to https://github.com/denoland/deno/pull/22226

cli/ops/testing.rs

@@ -1,6 +1,5 @@
 // Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
 
-use crate::tools::test::fmt::OP_DETAILS;
 use crate::tools::test::TestDescription;
 use crate::tools::test::TestEvent;
 use crate::tools::test::TestEventSender;
@@ -15,17 +14,11 @@ use deno_core::error::AnyError;
 use deno_core::op2;
 use deno_core::v8;
 use deno_core::ModuleSpecifier;
-use deno_core::OpMetricsSummary;
-use deno_core::OpMetricsSummaryTracker;
 use deno_core::OpState;
-use deno_runtime::deno_fetch::reqwest;
 use deno_runtime::permissions::create_child_permissions;
 use deno_runtime::permissions::ChildPermissionsArg;
 use deno_runtime::permissions::PermissionsContainer;
 use serde::Serialize;
-use std::collections::hash_map::Entry;
-use std::collections::HashMap;
-use std::rc::Rc;
 use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering;
 use uuid::Uuid;
@@ -45,10 +38,6 @@ deno_core::extension!(deno_test,
     op_test_event_step_result_ok,
     op_test_event_step_result_ignored,
     op_test_event_step_result_failed,
-    op_test_op_sanitizer_collect,
-    op_test_op_sanitizer_finish,
-    op_test_op_sanitizer_report,
-    op_test_op_sanitizer_get_async_message,
   ],
   options = {
     sender: TestEventSender,
@@ -56,7 +45,6 @@ deno_core::extension!(deno_test,
   state = |state, options| {
     state.put(options.sender);
     state.put(TestContainer::default());
-    state.put(TestOpSanitizers::default());
   },
 );
 
@@ -245,192 +233,3 @@ fn op_test_event_step_result_failed(
     ))
     .ok();
 }
-
-#[derive(Default)]
-struct TestOpSanitizers(HashMap<u32, TestOpSanitizerState>);
-
-enum TestOpSanitizerState {
-  Collecting { metrics: Vec<OpMetricsSummary> },
-  Finished { report: Vec<TestOpSanitizerReport> },
-}
-
-fn try_collect_metrics(
-  metrics: &OpMetricsSummaryTracker,
-  force: bool,
-  op_id_host_recv_msg: usize,
-  op_id_host_recv_ctrl: usize,
-) -> Result<std::cell::Ref<Vec<OpMetricsSummary>>, bool> {
-  let metrics = metrics.per_op();
-  let host_recv_msg = metrics
-    .get(op_id_host_recv_msg)
-    .map(OpMetricsSummary::has_outstanding_ops)
-    .unwrap_or(false);
-  let host_recv_ctrl = metrics
-    .get(op_id_host_recv_ctrl)
-    .map(OpMetricsSummary::has_outstanding_ops)
-    .unwrap_or(false);
-
-  for op_metric in metrics.iter() {
-    if op_metric.has_outstanding_ops() && !force {
-      return Err(host_recv_msg || host_recv_ctrl);
-    }
-  }
-  Ok(metrics)
-}
-
-#[op2(fast)]
-#[smi]
-// Returns:
-// 0 - success
-// 1 - for more accurate results, spin event loop and call again with force=true
-// 2 - for more accurate results, delay(1ms) and call again with force=true
-fn op_test_op_sanitizer_collect(
-  state: &mut OpState,
-  #[smi] id: u32,
-  force: bool,
-  #[smi] op_id_host_recv_msg: usize,
-  #[smi] op_id_host_recv_ctrl: usize,
-) -> Result<u8, AnyError> {
-  let metrics = state.borrow::<Rc<OpMetricsSummaryTracker>>();
-  let metrics = match try_collect_metrics(
-    metrics,
-    force,
-    op_id_host_recv_msg,
-    op_id_host_recv_ctrl,
-  ) {
-    Ok(metrics) => metrics,
-    Err(false) => {
-      return Ok(1);
-    }
-    Err(true) => {
-      return Ok(2);
-    }
-  }
-  .clone();
-
-  let op_sanitizers = state.borrow_mut::<TestOpSanitizers>();
-  match op_sanitizers.0.entry(id) {
-    Entry::Vacant(entry) => {
-      entry.insert(TestOpSanitizerState::Collecting { metrics });
-    }
-    Entry::Occupied(_) => {
-      return Err(generic_error(format!(
-        "Test metrics already being collected for test id {id}",
-      )));
-    }
-  }
-  Ok(0)
-}
-
-#[derive(Serialize)]
-#[serde(rename_all = "camelCase")]
-struct TestOpSanitizerReport {
-  id: usize,
-  diff: i64,
-}
-
-#[op2(fast)]
-#[smi]
-// Returns:
-// 0 - sanitizer finished with no pending ops
-// 1 - for more accurate results, spin event loop and call again with force=true
-// 2 - for more accurate results, delay(1ms) and call again with force=true
-// 3 - sanitizer finished with pending ops, collect the report with op_test_op_sanitizer_report
-fn op_test_op_sanitizer_finish(
-  state: &mut OpState,
-  #[smi] id: u32,
-  force: bool,
-  #[smi] op_id_host_recv_msg: usize,
-  #[smi] op_id_host_recv_ctrl: usize,
-) -> Result<u8, AnyError> {
-  // Drop `fetch` connection pool at the end of a test
-  state.try_take::<reqwest::Client>();
-  let metrics = state.borrow::<Rc<OpMetricsSummaryTracker>>();
-
-  // Generate a report of pending ops
-  let report = {
-    let after_metrics = match try_collect_metrics(
-      metrics,
-      force,
-      op_id_host_recv_msg,
-      op_id_host_recv_ctrl,
-    ) {
-      Ok(metrics) => metrics,
-      Err(false) => {
-        return Ok(1);
-      }
-      Err(true) => {
-        return Ok(2);
-      }
-    };
-
-    let op_sanitizers = state.borrow::<TestOpSanitizers>();
-    let before_metrics = match op_sanitizers.0.get(&id) {
-      Some(TestOpSanitizerState::Collecting { metrics }) => metrics,
-      _ => {
-        return Err(generic_error(format!(
-          "Metrics not collected before for test id {id}",
-        )));
-      }
-    };
-    let mut report = vec![];
-
-    for (id, (before, after)) in
-      before_metrics.iter().zip(after_metrics.iter()).enumerate()
-    {
-      let async_pending_before =
-        before.ops_dispatched_async - before.ops_completed_async;
-      let async_pending_after =
-        after.ops_dispatched_async - after.ops_completed_async;
-      let diff = async_pending_after as i64 - async_pending_before as i64;
-      if diff != 0 {
-        report.push(TestOpSanitizerReport { id, diff });
-      }
-    }
-
-    report
-  };
-
-  let op_sanitizers = state.borrow_mut::<TestOpSanitizers>();
-
-  if report.is_empty() {
-    op_sanitizers
-      .0
-      .remove(&id)
-      .expect("TestOpSanitizerState::Collecting");
-    Ok(0)
-  } else {
-    op_sanitizers
-      .0
-      .insert(id, TestOpSanitizerState::Finished { report })
-      .expect("TestOpSanitizerState::Collecting");
-    Ok(3)
-  }
-}
-
-#[op2]
-#[serde]
-fn op_test_op_sanitizer_report(
-  state: &mut OpState,
-  #[smi] id: u32,
-) -> Result<Vec<TestOpSanitizerReport>, AnyError> {
-  let op_sanitizers = state.borrow_mut::<TestOpSanitizers>();
-  match op_sanitizers.0.remove(&id) {
-    Some(TestOpSanitizerState::Finished { report }) => Ok(report),
-    _ => Err(generic_error(format!(
-      "Metrics not finished collecting for test id {id}",
-    ))),
-  }
-}
-
-#[op2]
-#[serde]
-fn op_test_op_sanitizer_get_async_message(
-  #[string] op_name: &str,
-) -> (String, Option<String>) {
-  if let Some(output) = OP_DETAILS.get(op_name) {
-    (output[0].to_string(), Some(output[1].to_string()))
-  } else {
-    (op_name.to_string(), None)
-  }
-}