Reapply "fix: remove actor system for transactions (#4940)" (#4975)

This reverts commit e1242e1.

libs/crosstarget-utils/src/native/time.rs

@@ -5,6 +5,7 @@ use std::{
 
 use crate::common::TimeoutError;
 
+#[derive(Clone, Copy)]
 pub struct ElapsedTimeCounter {
     instant: Instant,
 }

libs/crosstarget-utils/src/wasm/time.rs

@@ -21,6 +21,7 @@ extern "C" {
 
 }
 
+#[derive(Clone, Copy)]
 pub struct ElapsedTimeCounter {
     start_time: f64,
 }

libs/query-engine-common/Cargo.toml

@@ -8,6 +8,7 @@ thiserror = "1"
 url.workspace = true
 query-connector = { path = "../../query-engine/connectors/query-connector" }
 query-core = { path = "../../query-engine/core" }
+telemetry = { path = "../telemetry" }
 user-facing-errors = { path = "../user-facing-errors" }
 serde_json.workspace = true
 serde.workspace = true

libs/query-engine-common/src/tracer.rs

@@ -8,7 +8,6 @@ use opentelemetry::{
     },
     trace::{TraceError, TracerProvider},
 };
-use query_core::telemetry;
 use std::fmt::{self, Debug};
 
 /// Pipeline builder

libs/telemetry/Cargo.toml

@@ -0,0 +1,32 @@
+[package]
+edition = "2021"
+name = "telemetry"
+version = "0.1.0"
+
+[features]
+metrics = ["dep:query-engine-metrics"]
+
+[dependencies]
+async-trait.workspace = true
+crossbeam-channel = "0.5.6"
+psl.workspace = true
+futures = "0.3"
+indexmap.workspace = true
+itertools.workspace = true
+once_cell = "1"
+opentelemetry = { version = "0.17.0", features = ["rt-tokio", "serialize"] }
+serde.workspace = true
+serde_json.workspace = true
+thiserror = "1.0"
+tokio = { version = "1.0", features = ["macros", "time"] }
+tracing = { workspace = true, features = ["attributes"] }
+tracing-futures = "0.2"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+tracing-opentelemetry = "0.17.4"
+uuid.workspace = true
+cuid = { git = "https://github.com/prisma/cuid-rust", branch = "wasm32-support" }
+crosstarget-utils = { path = "../crosstarget-utils" }
+lru = "0.7.7"
+enumflags2.workspace = true
+derive_more = "0.99.17"
+query-engine-metrics = { path = "../../query-engine/metrics", optional = true }

query-engine/core/src/telemetry/capturing/capturer.rs → libs/telemetry/src/capturing/capturer.rs

@@ -29,6 +29,20 @@ impl Capturer {
 
         Self::Disabled
     }
+
+    pub async fn try_start_capturing(&self) {
+        if let Capturer::Enabled(capturer) = &self {
+            capturer.start_capturing().await
+        }
+    }
+
+    pub async fn try_fetch_captures(&self) -> Option<Storage> {
+        if let Capturer::Enabled(capturer) = &self {
+            capturer.fetch_captures().await
+        } else {
+            None
+        }
+    }
 }
 
 #[derive(Debug, Clone)]