From 4063071e0800a4feb21831980e5986de66465c9d Mon Sep 17 00:00:00 2001 From: Floriane TUERNAL SABOTINOV Date: Thu, 4 Sep 2025 14:10:16 +0200 Subject: [PATCH] Revert "broken: context won't satisfy spawn async move" This reverts commit 2b11479ad4801e1efa8ab0aca4ca577bd3f195fe. --- .../websocket/connection_handler/fresh.rs | 151 ++++++++++-------- .../websocket/connection_handler/mod.rs | 3 - 2 files changed, 88 insertions(+), 66 deletions(-) diff --git a/gateway/src/node/client_handling/websocket/connection_handler/fresh.rs b/gateway/src/node/client_handling/websocket/connection_handler/fresh.rs index 2ee79b6fc5..c9e243791b 100644 --- a/gateway/src/node/client_handling/websocket/connection_handler/fresh.rs +++ b/gateway/src/node/client_handling/websocket/connection_handler/fresh.rs @@ -858,77 +858,101 @@ impl FreshHandler { pub(crate) async fn handle_initial_client_request( &mut self, request: ClientControlRequest, - ) -> Result<(Option, Option), InitialAuthenticationError> + ) -> Result, InitialAuthenticationError> where S: AsyncRead + AsyncWrite + Unpin + Send, R: CryptoRng + RngCore + Send, { - let distributed_trace_id = if let ClientControlRequest::AuthenticateV2(ref auth_req) = request { - if let Some(ref trace_id_str) = auth_req.debug_trace_id { - warn!("Received debug trace ID: {trace_id_str}"); - match TraceId::from_hex(&trace_id_str) { - Ok(trace_id) => { - warn!("Parsed debug trace ID: {trace_id}"); - Some(trace_id) - } - Err(err) => { - warn!("Failed to parse debug trace ID: {err}"); - None - } - } + // let _distributed_context = if let ClientControlRequest::AuthenticateV2(ref auth_req) = request { + // if let Some(ref trace_id_str) = auth_req.debug_trace_id { + // warn!("Received debug trace ID: {trace_id_str}"); + + // match TraceId::from_hex(&trace_id_str) { + // Ok(trace_id) => { + // warn!("Parsed debug trace ID: {trace_id}"); + + // let id_gen = RandomIdGenerator::default(); + // let span_id = id_gen.new_span_id(); + + // let span_context = opentelemetry::trace::SpanContext::new( + // trace_id, + // span_id, + // opentelemetry::trace::TraceFlags::SAMPLED, + // true, + // Default::default(), + // ); + + // let remote_context = opentelemetry::Context::current() + // .with_remote_span_context(span_context); + + // let context_guard = remote_context.attach(); + + // let span = info_span!( + // "handle_initial_client_request", + // trace_id = %trace_id, + // service = "nym-node" + // ); + + // warn!("Distributed context established. trace_id: {:?}", trace_id); + // Some(span) + // } + // Err(err) => { + // warn!("Failed to parse debug trace ID: {err}"); + // None + // } + // } + // } else { + // warn!("Received debug trace ID without context"); + // None + // } + // } else { + // None + // }; + let span = if let ClientControlRequest::AuthenticateV2(ref auth_req) = request { + if let Some(ref trace_id) = auth_req.debug_trace_id { + warn!("RAW TRACE ID: {trace_id:?}"); + let trace_id = opentelemetry::trace::TraceId::from_hex(&trace_id) + .expect("Invalid trace ID format"); + warn!("🫂TraceID: {trace_id}🫂"); + // We don't need to try and preserve the SpanID, just the TraceID (right?) so + // just making a new SpanID for the moment + let id_generator = RandomIdGenerator::default(); + let span_id = id_generator.new_span_id(); + let span_context = opentelemetry::trace::SpanContext::new( + trace_id, + span_id, + opentelemetry::trace::TraceFlags::SAMPLED, + true, // is_remote = true since this comes from another service + Default::default(), + ); + let remote_context = + opentelemetry::Context::current().with_remote_span_context(span_context); + let contextguard = remote_context.clone().attach(); + let span = info_span!( + "authenticate_v2", + trace_id = %trace_id + ); + span.set_parent(remote_context.clone()); + Some(span) } else { warn!("AuthenticateV2 request but no trace_id provided"); None } } else { + warn!("Not an AuthenticateV2 request"); None }; - // mark/otel method - // let span = if let ClientControlRequest::AuthenticateV2(ref auth_req) = request { - // if let Some(ref trace_id) = auth_req.debug_trace_id { - // warn!("RAW TRACE ID: {trace_id:?}"); - // let trace_id = opentelemetry::trace::TraceId::from_hex(&trace_id) - // .expect("Invalid trace ID format"); - // warn!("🫂TraceID: {trace_id}🫂"); - // // We don't need to try and preserve the SpanID, just the TraceID (right?) so - // // just making a new SpanID for the moment - // let id_generator = RandomIdGenerator::default(); - // let span_id = id_generator.new_span_id(); - // let span_context = opentelemetry::trace::SpanContext::new( - // trace_id, - // span_id, - // opentelemetry::trace::TraceFlags::SAMPLED, - // true, // is_remote = true since this comes from another service - // Default::default(), - // ); - // let remote_context = - // opentelemetry::Context::current().with_remote_span_context(span_context); - // let contextguard = remote_context.clone().attach(); - // let span = info_span!( - // "authenticate_v2", - // trace_id = %trace_id - // ); - // span.set_parent(remote_context.clone()); - // Some(span) - // } else { - // warn!("AuthenticateV2 request but no trace_id provided"); - // None - // } - // } else { - // warn!("Not an AuthenticateV2 request"); - // None - // }; - // // Probably a nicer way to do this but for now just match - // let _guard = match &span { - // Some(s) => { - // warn!("ENTERED SPAN"); - // Some(s.enter()) - // } - // None => { - // warn!("COULDN'T ENTER SPAN"); - // None - // } - // }; + // Probably a nicer way to do this but for now just match + let _guard = match &span { + Some(s) => { + warn!("ENTERED SPAN"); + Some(s.enter()) + } + None => { + warn!("COULDN'T ENTER SPAN"); + None + } + }; let auth_result = match request { ClientControlRequest::Authenticate { @@ -948,7 +972,7 @@ impl FreshHandler { } => self.handle_register(protocol_version, data).await, ClientControlRequest::SupportedProtocol { .. } => { self.handle_reply_supported_protocol_request().await; - return Ok((None, distributed_trace_id)); + return Ok(None); } _ => { debug!("received an invalid client request"); @@ -988,7 +1012,7 @@ impl FreshHandler { return Err(InitialAuthenticationError::EmptyClientDetails); }; - Ok((Some(client_details), distributed_trace_id)) + Ok(Some(client_details)) } #[instrument(skip_all)] @@ -1010,7 +1034,7 @@ impl FreshHandler { }; // see if we managed to register the client through this request - let (maybe_auth_res, distributed_trace_id) = match self.handle_initial_client_request(initial_request).await { + let maybe_auth_res = match self.handle_initial_client_request(initial_request).await { Ok(maybe_auth_res) => maybe_auth_res, Err(err) => { debug!("initial client request handling error: {err}"); @@ -1021,6 +1045,7 @@ impl FreshHandler { if let Some(registration_details) = maybe_auth_res { let (mix_sender, mix_receiver) = mpsc::unbounded(); + // Channel for handlers to ask other handlers if they are still active. let (is_active_request_sender, is_active_request_receiver) = mpsc::unbounded(); self.shared_state.active_clients_store.insert_remote( registration_details.address, diff --git a/gateway/src/node/client_handling/websocket/connection_handler/mod.rs b/gateway/src/node/client_handling/websocket/connection_handler/mod.rs index e8a9ab2501..c8b59d4f1f 100644 --- a/gateway/src/node/client_handling/websocket/connection_handler/mod.rs +++ b/gateway/src/node/client_handling/websocket/connection_handler/mod.rs @@ -6,9 +6,6 @@ use nym_credential_verification::BandwidthFlushingBehaviourConfig; use nym_gateway_requests::shared_key::SharedGatewayKey; use nym_gateway_requests::ServerResponse; use nym_sphinx::DestinationAddressBytes; -use opentelemetry::Context; -use opentelemetry_sdk::trace::{IdGenerator, RandomIdGenerator}; -use opentelemetry::trace::TraceContextExt; use rand::{CryptoRng, Rng}; use std::time::Duration; use time::OffsetDateTime;