feat(coprocessor): migrate Phases C/D/H/J spans to tracing

Migrate remaining coprocessor workers from legacy OtelTracer to the
tracing + OTEL bridge pattern established in Phase A/B:

- Phase C: zkproof-worker verify_task spans (verifier.rs)
- Phase D: tfhe-worker high-signal execution loop spans (tfhe_worker.rs)
- Phase H: host-listener DB propagation spans (tfhe_event_propagate.rs)
- Phase J: transaction-sender op spans (add_ciphertext, allow_handle,
  delegate_user_decrypt, verify_proof)

Also cleans up telemetry.rs by removing the now-unused OtelTracer
wrapper and legacy span helpers, fixes a pre-existing clippy lint in
chain_id.rs, and fixes minor sns-worker span hygiene from Phase B.

Closes zama-ai/fhevm-internal#935
Closes zama-ai/fhevm-internal#1008
Closes zama-ai/fhevm-internal#1009
Closes zama-ai/fhevm-internal#1034
Closes zama-ai/fhevm-internal#1036

Author: Eikix

coprocessor/fhevm-engine/Cargo.lock

@@ -50,8 +50,8 @@ impl TryFrom<u64> for ChainId {
     type Error = InvalidChainId;
 
     fn try_from(value: u64) -> Result<Self, Self::Error> {
-        if value <= i64::MAX as u64 {
-            Ok(ChainId(value as i64))
+        if let Ok(v) = i64::try_from(value) {
+            Ok(ChainId(v))
         } else {
             Err(InvalidChainId {
                 value: value.to_string(),

coprocessor/fhevm-engine/fhevm-engine-common/src/chain_id.rs

@@ -1,11 +1,7 @@
 use crate::chain_id::ChainId;
 use crate::utils::to_hex;
 use bigdecimal::num_traits::ToPrimitive;
-use opentelemetry::{
-    global::{BoxedSpan, BoxedTracer, ObjectSafeSpan},
-    trace::{SpanBuilder, Status, TraceContextExt, Tracer, TracerProvider},
-    Context, KeyValue,
-};
+use opentelemetry::{global::BoxedSpan, trace::TracerProvider, KeyValue};
 use opentelemetry_sdk::{trace::SdkTracerProvider, Resource};
 use prometheus::{register_histogram, Histogram};
 use sqlx::PgConnection;
@@ -14,14 +10,25 @@ use std::{
     num::NonZeroUsize,
     str::FromStr,
     sync::{Arc, LazyLock, OnceLock},
-    time::SystemTime,
 };
 use tokio::sync::RwLock;
 use tracing::{debug, info, warn};
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 
 pub const TXN_ID_ATTR_KEY: &str = "txn_id";
 
+// Sets the txn_id attribute to the span
+// The txn_id is a shortened version of the transaction_id (first 10 characters of the hex representation)
+pub fn set_txn_id(span: &mut BoxedSpan, transaction_id: &[u8]) {
+    use opentelemetry::trace::Span;
+    let txn_id_short = to_hex(transaction_id)
+        .get(0..10)
+        .unwrap_or_default()
+        .to_owned();
+
+    span.set_attribute(KeyValue::new(TXN_ID_ATTR_KEY, txn_id_short));
+}
+
 /// Calls provider shutdown exactly once when dropped.
 pub struct TracerProviderGuard {
     tracer_provider: Option<SdkTracerProvider>,
@@ -142,119 +149,6 @@ fn setup_otel_with_tracer(
     Ok((tracer, trace_provider))
 }
 
-#[derive(Clone)]
-pub struct OtelTracer {
-    ctx: opentelemetry::Context,
-    tracer: Arc<BoxedTracer>,
-}
-
-impl OtelTracer {
-    pub fn child_span(&self, name: &'static str) -> BoxedSpan {
-        self.tracer.start_with_context(name, &self.ctx)
-    }
-
-    pub fn context(&self) -> &Context {
-        &self.ctx
-    }
-
-    /// Sets attribute to the root span
-    pub fn set_attribute(&self, key: &str, value: String) {
-        self.ctx
-            .span()
-            .set_attribute(KeyValue::new(key.to_owned(), value));
-    }
-
-    /// Consumes and ends the tracer with status Ok
-    pub fn end(self) {
-        self.ctx.span().set_status(Status::Ok);
-        self.ctx.span().end();
-    }
-}
-
-#[derive(Debug, PartialEq)]
-struct Handle(Vec<u8>);
-#[derive(Debug, PartialEq)]
-struct Transaction(Vec<u8>);
-
-pub fn tracer_with_handle(
-    span_name: &'static str,
-    handle: Vec<u8>,
-    transaction_id: &Option<Vec<u8>>,
-) -> OtelTracer {
-    let tracer = opentelemetry::global::tracer(format!("tracer_{}", span_name));
-    let mut span = tracer.start(span_name);
-
-    if !handle.is_empty() {
-        let handle = to_hex(&handle).get(0..10).unwrap_or_default().to_owned();
-
-        span.set_attribute(KeyValue::new("handle", handle));
-    }
-
-    if let Some(transaction_id) = transaction_id {
-        set_txn_id(&mut span, transaction_id);
-    }
-
-    // Add handle and transaction_id to the context
-    // so that they can be retrieved in the application code, e.g. for logging
-    let mut ctx = Context::default().with_span(span);
-    ctx = ctx.with_value(Handle(handle.clone()));
-    ctx = ctx.with_value(Transaction(transaction_id.clone().unwrap_or_default()));
-
-    OtelTracer {
-        ctx,
-        tracer: Arc::new(tracer),
-    }
-}
-
-// Sets the txn_id attribute to the span
-// The txn_id is a shortened version of the transaction_id (first 10 characters of the hex representation)
-pub fn set_txn_id(span: &mut BoxedSpan, transaction_id: &[u8]) {
-    let txn_id_short = to_hex(transaction_id)
-        .get(0..10)
-        .unwrap_or_default()
-        .to_owned();
-
-    span.set_attribute(KeyValue::new(TXN_ID_ATTR_KEY, txn_id_short));
-}
-
-/// Create a new span with start and end time
-pub fn tracer_with_start_time(span_name: &'static str, start_time: SystemTime) -> OtelTracer {
-    let tracer = opentelemetry::global::tracer(span_name);
-    let root_span = tracer.build(SpanBuilder::from_name(span_name).with_start_time(start_time));
-    let ctx = opentelemetry::Context::default().with_span(root_span);
-    OtelTracer {
-        ctx,
-        tracer: Arc::new(tracer),
-    }
-}
-
-pub fn tracer(span_name: &'static str, transaction_id: &Option<Vec<u8>>) -> OtelTracer {
-    tracer_with_handle(span_name, vec![], transaction_id)
-}
-
-pub fn attribute(span: &mut BoxedSpan, key: &str, value: String) {
-    span.set_attribute(KeyValue::new(key.to_owned(), value));
-}
-
-/// Ends span with status Ok
-pub fn end_span(mut span: BoxedSpan) {
-    span.set_status(Status::Ok);
-    span.end();
-}
-
-pub fn end_span_with_timestamp(mut span: BoxedSpan, timestamp: SystemTime) {
-    span.set_status(Status::Ok);
-    span.end_with_timestamp(timestamp);
-}
-
-/// Ends span with status Error with description
-pub fn end_span_with_err(mut span: BoxedSpan, desc: String) {
-    span.set_status(Status::Error {
-        description: desc.into(),
-    });
-    span.end();
-}
-
 #[derive(Clone, Copy, Debug)]
 pub struct MetricsConfig {
     bucket_start: f64,

coprocessor/fhevm-engine/fhevm-engine-common/src/telemetry.rs

@@ -295,10 +295,8 @@ impl Database {
             self.insert_computation_bytes(tx, result, dependencies_handles, dependencies_bytes, fhe_operation, scalar_byte, log)
         };
 
-        let _t = telemetry::tracer(
-            "handle_tfhe_event",
-            &log.transaction_hash.map(|h| h.to_vec()),
-        );
+        let _span = tracing::info_span!("handle_tfhe_event", operation = "handle_tfhe_event");
+        let _enter = _span.enter();
 
         // Record the transaction if this is a computation event
         if !matches!(
@@ -453,7 +451,11 @@ impl Database {
 
         let transaction_hash = transaction_hash.map(|h| h.to_vec());
 
-        let _t = telemetry::tracer("handle_acl_event", &transaction_hash);
+        let _span = tracing::info_span!(
+            "handle_acl_event",
+            operation = "handle_acl_event"
+        );
+        let _enter = _span.enter();
 
         // Record only Allowed or AllowedForDecryption events
         if matches!(

coprocessor/fhevm-engine/host-listener/src/database/tfhe_event_propagate.rs

@@ -10,7 +10,6 @@ use aws_sdk_s3::Client;
 use bytesize::ByteSize;
 use fhevm_engine_common::chain_id::ChainId;
 use fhevm_engine_common::pg_pool::{PostgresPoolManager, ServiceError};
-use fhevm_engine_common::telemetry;
 use fhevm_engine_common::utils::to_hex;
 use futures::future::join_all;
 use opentelemetry::trace::{Status, TraceContextExt};
@@ -166,7 +165,7 @@ async fn run_uploader_loop(
                 // Spawn a new task to upload the ciphertexts
                 let h = tokio::spawn(async move {
                     let upload_span = error_span!("upload_s3", operation = "upload_s3");
-                    upload_span.set_parent(item.otel.context().clone());
+                    upload_span.set_parent(item.otel.context());
                     match upload_ciphertexts(trx, item, &client, &conf)
                         .instrument(upload_span.clone())
                         .await
@@ -543,7 +542,7 @@ async fn fetch_pending_uploads(
                 handle: handle.clone(),
                 ct64_compressed,
                 ct128: Arc::new(ct128),
-                otel: telemetry::tracer_with_handle("recovery_task", handle, &transaction_id),
+                otel: tracing::info_span!("recovery_task", operation = "recovery_task"),
                 transaction_id,
             };
 

coprocessor/fhevm-engine/sns-worker/src/aws_upload.rs

@@ -180,7 +180,7 @@ async fn get_keyset(
     pool: PgPool,
     keys_cache: Arc<RwLock<lru::LruCache<DbKeyId, KeySet>>>,
 ) -> Result<Option<(DbKeyId, KeySet)>, ExecutionError> {
-    let _t = telemetry::tracer("fetch_keyset", &None);
+    let _t = tracing::info_span!("fetch_keyset", operation = "fetch_keyset");
     fetch_latest_keyset(&keys_cache, &pool).await
 }
 
@@ -310,7 +310,7 @@ pub async fn garbage_collect(pool: &PgPool, limit: u32) -> Result<(), ExecutionE
 
     // Limit the number of rows to update in case of a large backlog due to catchup or burst
     // Skip Locked to prevent concurrent updates
-    let start = SystemTime::now();
+    let _start = SystemTime::now();
     let rows_affected: u64 = sqlx::query!(
         "
         WITH uploaded_ct128 AS (
@@ -334,7 +334,7 @@ pub async fn garbage_collect(pool: &PgPool, limit: u32) -> Result<(), ExecutionE
     .rows_affected();
 
     if rows_affected > 0 {
-        let _s = telemetry::tracer_with_start_time("cleanup_ct128", start);
+        let _s = tracing::info_span!("cleanup_ct128", operation = "cleanup_ct128");
         info!(
             rows_affected = rows_affected,
             "Cleaning up old ciphertexts128"
@@ -377,8 +377,11 @@ async fn fetch_and_execute_sns_tasks(
         maybe_remaining = conf.db.batch_limit as usize == tasks.len();
         tasks_processed = tasks.len();
 
-        let t = telemetry::tracer("batch_execution", &None);
-        t.set_attribute("count", tasks.len().to_string());
+        let t = tracing::info_span!(
+            "batch_execution",
+            operation = "batch_execution",
+            count = tasks.len()
+        );
 
         process_tasks(
             &mut tasks,
@@ -395,7 +398,7 @@ async fn fetch_and_execute_sns_tasks(
             "batch_store_ciphertext128",
             operation = "batch_store_ciphertext128"
         );
-        batch_store_span.set_parent(t.context().clone());
+        batch_store_span.set_parent(t.context());
         let batch_store = async {
             update_ciphertext128(trx, &tasks).await?;
             notify_ciphertext128_ready(trx, &conf.db.notify_channel).await?;
@@ -436,7 +439,7 @@ pub async fn query_sns_tasks(
     order: Order,
     key_id_gw: &DbKeyId,
 ) -> Result<Option<Vec<HandleItem>>, ExecutionError> {
-    let start_time = SystemTime::now();
+    let _start_time = SystemTime::now();
 
     let query = format!(
         "
@@ -464,9 +467,13 @@ pub async fn query_sns_tasks(
         return Ok(None);
     }
 
-    let t = telemetry::tracer_with_start_time("db_fetch_tasks", start_time);
-    t.set_attribute("count", records.len().to_string());
-    t.end();
+    {
+        let _t = tracing::info_span!(
+            "db_fetch_tasks",
+            operation = "db_fetch_tasks",
+            count = records.len()
+        );
+    }
 
     // Convert the records into HandleItem structs
     let tasks = records
@@ -487,7 +494,7 @@ pub async fn query_sns_tasks(
                 handle: handle.clone(),
                 ct64_compressed: Arc::new(ciphertext),
                 ct128: Arc::new(BigCiphertext::default()), // to be computed
-                otel: telemetry::tracer_with_handle("task", handle, &transaction_id),
+                otel: tracing::info_span!("task", operation = "task"),
                 transaction_id,
             })
         })
@@ -563,7 +570,7 @@ fn compute_task(
     let started_at = SystemTime::now();
     let thread_id = format!("{:?}", std::thread::current().id());
     let span = error_span!("compute", thread_id = %thread_id);
-    span.set_parent(task.otel.context().clone());
+    span.set_parent(task.otel.context());
     let _enter = span.enter();
 
     let handle = to_hex(&task.handle);

coprocessor/fhevm-engine/sns-worker/src/executor.rs

@@ -24,7 +24,6 @@ use fhevm_engine_common::{
     healthz_server::{self},
     metrics_server,
     pg_pool::{PostgresPoolManager, ServiceError},
-    telemetry::OtelTracer,
     types::FhevmError,
     utils::{to_hex, DatabaseURL},
 };
@@ -237,7 +236,7 @@ pub struct HandleItem {
     /// The computed 128-bit ciphertext
     pub(crate) ct128: Arc<BigCiphertext>,
 
-    pub otel: OtelTracer,
+    pub otel: tracing::Span,
     pub transaction_id: Option<Vec<u8>>,
 }
 

coprocessor/fhevm-engine/sns-worker/src/lib.rs

@@ -28,6 +28,7 @@ tokio = { workspace = true }
 tokio-util = { workspace = true }
 tonic = { workspace = true }
 tracing = { workspace = true }
+tracing-opentelemetry = { workspace = true }
 tracing-subscriber = { workspace = true }
 # opentelemetry support
 opentelemetry = { workspace = true }