GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

continue trace)id propagation into sphinx

Browse Source
This commit is contained in:
Floriane TUERNAL SABOTINOV
2025-10-08 16:14:05 +02:00
parent 9a96be3d41
commit b292c4a624
18 changed files with 315 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+2
View File
@@ -6981,6 +6981,7 @@ name = "nym-sphinx-addressing"
version = "0.1.0"
dependencies = [
"bincode",
"nym-bin-common",
"nym-crypto",
"nym-sphinx-types",
"rand 0.8.5",
@@ -7064,6 +7065,7 @@ dependencies = [
"nym-sphinx-forwarding",
"nym-sphinx-params",
"nym-sphinx-types",
"opentelemetry",
"thiserror 2.0.17",
"tokio",
"tokio-util",
common/bin-common/src/opentelemetry/compact_id_generator.rs
+3 -3
View File
@@ -39,9 +39,9 @@ pub fn compress_trace_id(trace_id: &TraceId) -> [u8; 12] {
compressed
}
pub fn decompress_trace_id(compressed: &[u8; 12]) -> TraceId {
pub fn decompress_trace_id(compressed: &[u8; 12]) -> [u8; 16] {
let mut bytes = [0u8; 16];
bytes[0..12].copy_from_slice(compressed);
TraceId::from_bytes(bytes)
bytes[12..].copy_from_slice(&[0u8; 4]);
bytes
}
common/bin-common/src/opentelemetry/mod.rs
+2 -2
View File
@@ -74,7 +74,7 @@ pub fn setup_tracing_logger(service_name: String) -> Result<(), TracingError> {
.with(console_layer)
.with(fmt_layer)
.with(granual_filtered_env()?)
.with(tracing_subscriber::filter::LevelFilter::from_level(Level::DEBUG))
.with(tracing_subscriber::filter::LevelFilter::from_level(Level::WARN))
.with(MetricsLayer::new(meter_provider))
.with(OpenTelemetryLayer::new(tracer))
.try_init()
@@ -83,7 +83,7 @@ pub fn setup_tracing_logger(service_name: String) -> Result<(), TracingError> {
tracing_subscriber::registry()
.with(fmt_layer)
.with(granual_filtered_env()?)
.with(tracing_subscriber::filter::LevelFilter::from_level(Level::DEBUG))
.with(tracing_subscriber::filter::LevelFilter::from_level(Level::WARN))
.with(MetricsLayer::new(meter_provider))
.with(OpenTelemetryLayer::new(tracer))
.try_init()
common/client-core/src/client/real_messages_control/acknowledgement_control/input_message_listener.rs
+12
View File
@@ -96,6 +96,7 @@ where
lane: TransmissionLane,
packet_type: PacketType,
max_retransmissions: Option<u32>,
trace_id: Option<[u8; 12]>,
) {
if let Err(err) = self
.message_handler
@@ -106,6 +107,7 @@ where
lane,
packet_type,
max_retransmissions,
trace_id,
)
.await
{
@@ -114,8 +116,12 @@ where
}
#[allow(clippy::panic)]
#[instrument(skip_all)]
async fn on_input_message(&mut self, msg: InputMessage) {
let trace_id = msg.trace_id();
if let Some(tid) = trace_id {
tracing::warn!("Processing input message with trace_id: {:?}", tid);
}
match msg {
InputMessage::Regular {
@@ -125,6 +131,7 @@ where
max_retransmissions,
..
} => {
warn!("Handling regular input message with trace_id: {:?}", trace_id);
self.handle_plain_message(
recipient,
data,
@@ -143,6 +150,7 @@ where
max_retransmissions,
..
} => {
warn!("Handling anonymous input message with trace_id: {:?}", trace_id);
self.handle_repliable_message(
recipient,
data,
@@ -150,6 +158,7 @@ where
lane,
PacketType::Mix,
max_retransmissions,
trace_id
)
.await
}
@@ -159,6 +168,7 @@ where
lane,
max_retransmissions,
} => {
warn!("Handling reply input message with trace_id: {:?}", trace_id);
self.handle_reply(recipient_tag, data, lane, max_retransmissions)
.await;
}
@@ -174,6 +184,7 @@ where
max_retransmissions,
..
} => {
tracing::warn!("Handling regular input message with trace_id: {:?}", trace_id);
self.handle_plain_message(
recipient,
data,
@@ -199,6 +210,7 @@ where
lane,
packet_type,
max_retransmissions,
trace_id
)
.await
}
common/client-core/src/client/real_messages_control/message_handler.rs
+2 -1
View File
@@ -607,6 +607,7 @@ where
lane: TransmissionLane,
packet_type: PacketType,
max_retransmissions: Option<u32>,
trace_id: Option<[u8; 12]>,
) -> Result<(), SurbWrappedPreparationError> {
debug!("Sending message with reply SURBs with packet type {packet_type}");
let sender_tag = self.get_or_create_sender_tag(&recipient);
@@ -630,7 +631,7 @@ where
lane,
packet_type,
max_retransmissions,
None,
trace_id,
)
.await?;
common/nymsphinx/acknowledgements/src/surb_ack.rs
+1 -1
View File
@@ -61,7 +61,7 @@ impl SurbAck {
};
let delays = nym_sphinx_routing::generate_hop_delays(average_delay, route.len());
let destination = recipient.as_sphinx_destination();
let destination = recipient.as_sphinx_destination(None);
let surb_ack_payload = prepare_identifier(rng, ack_key, marshaled_fragment_id);
let packet_size = match packet_type {
common/nymsphinx/addressing/Cargo.toml
+1
View File
@@ -8,6 +8,7 @@ license = { workspace = true }
repository = { workspace = true }
[dependencies]
nym-bin-common = { path = "../../bin-common", features = ["opentelemetry"] } # for trace id compression/decompression
nym-crypto = { path = "../../crypto", features = ["asymmetric", "sphinx"] } # all addresses are expressed in terms on their crypto keys
nym-sphinx-types = { path = "../types", features = ["sphinx"] } # we need to be able to refer to some types defined inside sphinx crate
serde = { workspace = true } # implementing serialization/deserialization for some types, like `Recipient`
common/nymsphinx/addressing/src/clients.rs
+9 -2
View File
@@ -153,12 +153,19 @@ impl Recipient {
// TODO: Currently the `DestinationAddress` is equivalent to `ClientIdentity`, but perhaps
// it shouldn't be? Maybe it should be (for example) H(`ClientIdentity || ClientEncryptionKey`)
// instead? That is an open question.
pub fn as_sphinx_destination(&self) -> Destination {
pub fn as_sphinx_destination(&self, trace_id: Option<[u8; 12]>) -> Destination {
use nym_bin_common::opentelemetry::compact_id_generator::decompress_trace_id;
let trace_id_16 = if let Some(trace_id) = trace_id {
decompress_trace_id(&trace_id)
} else {
decompress_trace_id(&[0u8; 12])
};
// since the nym mix network differs slightly in design from loopix, we do not care
// about "surb_id" field at all and just use the default value.
Destination::new(
self.client_identity.derive_destination_address(),
Default::default(),
trace_id_16,
)
}
common/nymsphinx/anonymous-replies/src/reply_surb.rs
+1 -1
View File
@@ -82,7 +82,7 @@ impl ReplySurb {
topology.random_route_to_egress(rng, recipient.gateway())?
};
let delays = nym_sphinx_routing::generate_hop_delays(average_delay, route.len());
let destination = recipient.as_sphinx_destination();
let destination = recipient.as_sphinx_destination(None);
let mut surb_material = SURBMaterial::new(route, delays, destination);
if use_legacy_surb_format && !disable_mix_hops {
common/nymsphinx/cover/src/lib.rs
+1 -1
View File
@@ -125,7 +125,7 @@ where
let route = topology.random_route_to_egress(rng, full_address.gateway())?;
let delays = nym_sphinx_routing::generate_hop_delays(average_packet_delay, route.len());
let destination = full_address.as_sphinx_destination();
let destination = full_address.as_sphinx_destination(None);
let rotation_id = topology.current_key_rotation();
let sphinx_key_rotation = SphinxKeyRotation::from(rotation_id);
common/nymsphinx/framing/Cargo.toml
+1
View File
@@ -11,6 +11,7 @@ repository = { workspace = true }
bytes = { workspace = true }
tokio-util = { workspace = true, features = ["codec"] }
thiserror = { workspace = true }
opentelemetry = { workspace = true }
tracing = { workspace = true }
nym-bin-common = { path = "../../bin-common" }
common/nymsphinx/framing/src/processing.rs
+8 -3
View File
@@ -15,7 +15,7 @@ use nym_sphinx_types::{
};
use std::fmt::Display;
use thiserror::Error;
use tracing::{debug, error, info, trace};
use tracing::{debug, error, info, warn, trace, instrument};
#[derive(Debug)]
pub enum MixProcessingResultData {
@@ -237,6 +237,7 @@ fn perform_framed_packet_processing(
})
}
#[instrument(skip_all)]
fn wrap_processed_sphinx_packet(
packet: nym_sphinx_types::ProcessedPacket,
packet_size: PacketSize,
@@ -267,9 +268,13 @@ fn wrap_processed_sphinx_packet(
{
let trace_bytes: [u8; 12] = identifier[0..12].try_into().unwrap();
if !trace_bytes.iter().all(|b| *b == 0) {
let full_trace_id = decompress_trace_id(&trace_bytes);
let full_trace_id_bytes = decompress_trace_id(&trace_bytes);
let full_trace_id = opentelemetry::trace::TraceId::from_bytes(full_trace_id_bytes);
tracing::Span::current().record("trace_id", &full_trace_id.to_string().as_str());
info!("Processing final hop with trace_id: {full_trace_id:?}");
warn!("Processing final hop with trace_id: {full_trace_id:?}");
}
else {
warn!("Received final hop with empty trace_id");
}
}
common/nymsphinx/src/preparer/mod.rs
+4 -15
View File
@@ -11,8 +11,7 @@ use nym_sphinx_addressing::clients::Recipient;
use nym_sphinx_addressing::nodes::NymNodeRoutingAddress;
use nym_sphinx_anonymous_replies::reply_surb::ReplySurb;
use nym_sphinx_chunking::fragment::{Fragment, FragmentIdentifier};
use nym_sphinx_forwarding::packet::{self, MixPacket};
use sphinx_packet::route::Destination;
use nym_sphinx_forwarding::packet::MixPacket;
use nym_sphinx_params::packet_sizes::PacketSize;
use nym_sphinx_params::{PacketType, ReplySurbKeyDigestAlgorithm, SphinxKeyRotation};
use nym_sphinx_types::{Delay, NymPacket};
@@ -169,6 +168,7 @@ pub trait FragmentPreparer {
/// - compute sphinx_plaintext = SURB_ACK || g^x || v_b
/// - compute sphinx_packet = Sphinx(recipient, sphinx_plaintext)
#[allow(clippy::too_many_arguments)]
#[instrument(skip_all)]
fn prepare_chunk_for_sending(
&mut self,
fragment: Fragment,
@@ -238,8 +238,8 @@ pub trait FragmentPreparer {
topology.random_route_to_egress(&mut rng, destination)?
};
let mut destination = packet_recipient.as_sphinx_destination();
add_trace_id_to_destination(&mut destination, trace_id);
let destination = packet_recipient.as_sphinx_destination(trace_id);
tracing::warn!("Packet destination with trace id: {:?}", &destination.identifier);
// including set of delays
let delays =
@@ -264,17 +264,6 @@ pub trait FragmentPreparer {
)?,
};
/// If a trace id is provided, add it to the first 12 bytes of the destination identifier.
/// The remaining 4 bytes of the identifier are left unchanged.
fn add_trace_id_to_destination(
destination: &mut Destination,
trace_id: Option<[u8; 12]>,
) {
if let Some(trace_id) = trace_id {
destination.identifier[0..12].copy_from_slice(&trace_id);
}
}
/// - compute k = KDF(remote encryption key ^ x) this is equivalent to KDF( dh(remote, x) )
/// from the previously constructed route extract the first hop
let first_hop_address =
gateway/src/node/client_handling/websocket/connection_handler/fresh.rs
+1
View File
@@ -872,6 +872,7 @@ impl<R, S> FreshHandler<R, S> {
// extract and set up opentelemetry context if provided
let (context_propagator, otel_ctx) = if let ClientControlRequest::AuthenticateV2(ref auth_req) = request {
if let Some(otel_context) = &auth_req.otel_context {
warn!("OpenTelemetry context provided in the request: {otel_context:?}");
(Some(ManualContextPropagator::new("handling_initial_client_request_with_otel", otel_context.clone())), Some(otel_context.clone()))
} else {
warn!("No OpenTelemetry context provided in the request");
nym-node/src/config/mod.rs
+14 -14
View File
@@ -428,20 +428,20 @@ impl Config {
pub fn validate(&self) -> Result<(), NymNodeError> {
self.mixnet.validate()?;
// it's not allowed to run mixnode mode alongside entry mode
if self.modes.mixnode && self.modes.entry {
return Err(NymNodeError::config_validation_failure(
"illegal modes configuration - node cannot run as a mixnode and an entry gateway",
));
}
// nor it's allowed to run mixnode mode alongside exit mode
// (use two separate checks for better error messages)
if self.modes.mixnode && self.modes.exit {
return Err(NymNodeError::config_validation_failure(
"illegal modes configuration - node cannot run as a mixnode and an exit gateway",
));
}
// // it's not allowed to run mixnode mode alongside entry mode
// if self.modes.mixnode && self.modes.entry {
// return Err(NymNodeError::config_validation_failure(
// "illegal modes configuration - node cannot run as a mixnode and an entry gateway",
// ));
// }
//
// // nor it's allowed to run mixnode mode alongside exit mode
// // (use two separate checks for better error messages)
// if self.modes.mixnode && self.modes.exit {
// return Err(NymNodeError::config_validation_failure(
// "illegal modes configuration - node cannot run as a mixnode and an exit gateway",
// ));
// }
Ok(())
}
sdk/rust/nym-sdk/examples/custom_topology_provider.rs
+144 -77
View File
@@ -1,102 +1,169 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use nym_sdk::mixnet;
use futures::StreamExt;
use nym_sdk::mixnet::MixnetMessageSender;
use nym_sdk::{mixnet, DebugConfig};
use nym_topology::provider_trait::{async_trait, ToTopologyMetadata, TopologyProvider};
use nym_topology::{EpochRewardedSet, NymTopology};
use nym_topology::{
CachedEpochRewardedSet, EntryDetails, EpochRewardedSet, HardcodedTopologyProvider, NymTopology,
NymTopologyMetadata, RoutingNode, SupportedRoles,
};
use nym_validator_client::nym_api::NymApiClientExt;
use std::time::Duration;
use time::OffsetDateTime;
use tokio::time::sleep;
use url::Url;
struct MyTopologyProvider {
validator_client: nym_http_api_client::Client,
}
impl MyTopologyProvider {
fn new(nym_api_url: Url) -> MyTopologyProvider {
let validator_client = nym_http_api_client::Client::builder(nym_api_url)
.expect("Failed to create API client builder")
.build()
.expect("Failed to build API client");
MyTopologyProvider { validator_client }
}
async fn get_topology(&self) -> NymTopology {
let rewarded_set = self
.validator_client
.get_current_rewarded_set()
.await
.unwrap();
let mixnodes_response = self
.validator_client
.get_all_basic_active_mixing_assigned_nodes_with_metadata()
.await
.unwrap();
let metadata = mixnodes_response.metadata.to_topology_metadata();
let epoch_rewarded_set: EpochRewardedSet = rewarded_set.into();
let mut base_topology = NymTopology::new(metadata, epoch_rewarded_set, Vec::new());
// in our topology provider only use mixnodes that have node_id divisible by 3
// and has exactly 100 performance score
// why? because this is just an example to showcase arbitrary uses and capabilities of this trait
let filtered_mixnodes = mixnodes_response
.nodes
.into_iter()
.filter(|mix| mix.node_id % 3 == 0 && mix.performance.is_hundred())
.collect::<Vec<_>>();
let gateways = self
.validator_client
.get_all_basic_entry_assigned_nodes_with_metadata()
.await
.unwrap()
.nodes;
base_topology.add_skimmed_nodes(&filtered_mixnodes);
base_topology.add_skimmed_nodes(&gateways);
base_topology
}
}
#[async_trait]
impl TopologyProvider for MyTopologyProvider {
// this will be manually refreshed on a timer specified inside mixnet client config
async fn get_new_topology(&mut self) -> Option<NymTopology> {
Some(self.get_topology().await)
}
}
//
// struct MyTopologyProvider {
// validator_client: nym_http_api_client::Client,
// }
//
// impl MyTopologyProvider {
// fn new(nym_api_url: Url) -> MyTopologyProvider {
// let validator_client = nym_http_api_client::Client::builder(nym_api_url)
// .expect("Failed to create API client builder")
// .build()
// .expect("Failed to build API client");
//
// MyTopologyProvider { validator_client }
// }
//
// async fn get_topology(&self) -> NymTopology {
// let rewarded_set = self
// .validator_client
// .get_current_rewarded_set()
// .await
// .unwrap();
//
// let mixnodes_response = self
// .validator_client
// .get_all_basic_active_mixing_assigned_nodes_with_metadata()
// .await
// .unwrap();
//
// let metadata = mixnodes_response.metadata.to_topology_metadata();
//
// let epoch_rewarded_set: EpochRewardedSet = rewarded_set.into();
// let mut base_topology = NymTopology::new(metadata, epoch_rewarded_set, Vec::new());
//
// // in our topology provider only use mixnodes that have node_id divisible by 3
// // and has exactly 100 performance score
// // why? because this is just an example to showcase arbitrary uses and capabilities of this trait
// let filtered_mixnodes = mixnodes_response
// .nodes
// .into_iter()
// .filter(|mix| mix.node_id % 3 == 0 && mix.performance.is_hundred())
// .collect::<Vec<_>>();
//
// let gateways = self
// .validator_client
// .get_all_basic_entry_assigned_nodes_with_metadata()
// .await
// .unwrap()
// .nodes;
//
// base_topology.add_skimmed_nodes(&filtered_mixnodes);
// base_topology.add_skimmed_nodes(&gateways);
// base_topology
// }
// }
//
// #[async_trait]
// impl TopologyProvider for MyTopologyProvider {
// // this will be manually refreshed on a timer specified inside mixnet client config
// async fn get_new_topology(&mut self) -> Option<NymTopology> {
// Some(self.get_topology().await)
// }
// }
#[tokio::main]
async fn main() {
nym_bin_common::logging::setup_tracing_logger();
let nym_api = "https://validator.nymtech.net/api/".parse().unwrap();
let my_topology_provider = MyTopologyProvider::new(nym_api);
// let nym_api = "https://validator.nymtech.net/api/".parse().unwrap();
// let my_topology_provider = MyTopologyProvider::new(nym_api);
let topology_metadata = NymTopologyMetadata::new(0, 0, OffsetDateTime::now_utc());
let mut rewarded_set = CachedEpochRewardedSet::default();
rewarded_set.entry_gateways.insert(1);
rewarded_set.layer1.insert(1);
rewarded_set.layer2.insert(1);
rewarded_set.layer3.insert(1);
let nodes = vec![RoutingNode {
node_id: 1,
mix_host: "127.0.0.1:1789".parse().unwrap(),
entry: Some(EntryDetails {
ip_addresses: vec!["127.0.0.1".parse().unwrap()],
clients_ws_port: 9000,
hostname: None,
clients_wss_port: None,
}),
identity_key: "PUT IDENTITY KEY HERE"
.parse()
.unwrap(),
sphinx_key: "PUT SPHINX KEY HERE"
.parse()
.unwrap(),
supported_roles: SupportedRoles {
mixnode: true,
mixnet_entry: true,
mixnet_exit: true,
},
}];
let topology_provider =
HardcodedTopologyProvider::new(NymTopology::new(topology_metadata, rewarded_set, nodes));
// Passing no config makes the client fire up an ephemeral session and figure things out on its own
let mut client = mixnet::MixnetClientBuilder::new_ephemeral()
.custom_topology_provider(Box::new(my_topology_provider))
.custom_topology_provider(Box::new(topology_provider))
.build()
.unwrap()
.connect_to_mixnet()
.await
.unwrap();
let our_address = client.nym_address();
let our_address = *client.nym_address();
println!("Our client nym address is: {our_address}");
// Send a message through the mixnet to ourselves
client
.send_plain_message(*our_address, "hello there")
.await
.unwrap();
let sender = client.split_sender();
println!("Waiting for message (ctrl-c to exit)");
client
.on_messages(|msg| println!("Received: {}", String::from_utf8_lossy(&msg.message)))
.await;
const MAX_MESSAGES: usize = 100;
// receiving task
let receiving_task_handle = tokio::spawn(async move {
let mut received_count = 0;
while let Some(received) = client.next().await {
received_count += 1;
println!(
"{received_count}: received: {}",
String::from_utf8_lossy(&received.message)
);
if received_count >= MAX_MESSAGES {
break;
}
}
client.disconnect().await;
});
// sending task
let sending_task_handle = tokio::spawn(async move {
loop {
if sender
.send_plain_message(our_address, "hello there")
.await
.is_err()
{
break;
}
sleep(Duration::from_millis(100)).await;
}
});
// wait for both tasks to be done
println!("waiting for shutdown");
sending_task_handle.await.unwrap();
receiving_task_handle.await.unwrap();
}
sdk/rust/nym-sdk/examples/local_surb_example.rs
+108
View File
@@ -0,0 +1,108 @@
use nym_sdk::mixnet::{
AnonymousSenderTag, MixnetClientBuilder, MixnetMessageSender, ReconstructedMessage,
};
use nym_topology::{CachedEpochRewardedSet, EntryDetails, HardcodedTopologyProvider, NymTopology, NymTopologyMetadata, RoutingNode, SupportedRoles};
use opentelemetry::trace::{TraceContextExt, Tracer};
use opentelemetry::{global, Context};
use time::OffsetDateTime;
use tracing::warn;
use tracing::instrument;
#[tokio::main]
#[instrument(name = "sdk-example-surb-reply", skip_all)]
async fn main() {
nym_bin_common::opentelemetry::setup_tracing_logger("local-sdk-example-surb-reply".to_string()).unwrap();
let tracer = global::tracer("local-sdk-example-surb-reply");
let span = tracer.start("client-root-span");
let cx = Context::current_with_span(span);
let _guard = cx.clone().attach();
let trace_id = cx.span().span_context().trace_id();
warn!("Main TRACE_ID: {:?}", trace_id);
// Create a mixnet client which connect to a local node
let topology_metadata = NymTopologyMetadata::new(0, 0, OffsetDateTime::now_utc());
let mut rewarded_set = CachedEpochRewardedSet::default();
rewarded_set.entry_gateways.insert(1);
rewarded_set.layer1.insert(1);
rewarded_set.layer2.insert(1);
rewarded_set.layer3.insert(1);
let nodes = vec![RoutingNode {
node_id: 1,
mix_host: "127.0.0.1:1789".parse().unwrap(),
entry: Some(EntryDetails {
ip_addresses: vec!["127.0.0.1".parse().unwrap()],
clients_ws_port: 9000,
hostname: None,
clients_wss_port: None,
}),
identity_key: "Put Your Identity Key Here"
.parse()
.unwrap(),
sphinx_key: "Put Your Sphinx Key Here"
.parse()
.unwrap(),
supported_roles: SupportedRoles {
mixnode: true,
mixnet_entry: true,
mixnet_exit: true,
},
}];
let topology_provider =
HardcodedTopologyProvider::new(NymTopology::new(topology_metadata, rewarded_set, nodes));
let mut client = MixnetClientBuilder::new_ephemeral()
.custom_topology_provider(Box::new(topology_provider))
.build().unwrap()
.connect_to_mixnet()
.await
.unwrap();
let our_address = client.nym_address();
println!("\nOur client nym address is: {our_address}");
// Send a message through the mixnet to ourselves using our nym address
client
.send_plain_message(*our_address, "hello there")
.await
.unwrap();
// we're going to parse the sender_tag (AnonymousSenderTag) from the incoming message and use it to 'reply' to ourselves instead of our Nym address.
// we know there will be a sender_tag since the sdk sends SURBs along with messages by default.
println!("Waiting for message\n");
// get the actual message - discard the empty vec sent along with a potential SURB topup request
let mut message: Vec<ReconstructedMessage> = Vec::new();
while let Some(new_message) = client.wait_for_messages().await {
if new_message.is_empty() {
continue;
}
message = new_message;
break;
}
let mut parsed = String::new();
if let Some(r) = message.first() {
parsed = String::from_utf8(r.message.clone()).unwrap();
}
// parse sender_tag: we will use this to reply to sender without needing their Nym address
let return_recipient: AnonymousSenderTag = message[0].sender_tag.unwrap();
println!(
"\nReceived the following message: {parsed} \nfrom sender with surb bucket {return_recipient}"
);
// reply to self with it: note we use `send_str_reply` instead of `send_str`
println!("Replying with using SURBs");
client
.send_reply(return_recipient, "hi an0n!")
.await
.unwrap();
println!("Waiting for message (once you see it, ctrl-c to exit)\n");
client
.on_messages(|msg| println!("\nReceived: {}", String::from_utf8_lossy(&msg.message)))
.await;
}
sdk/rust/nym-sdk/src/mixnet/traits.rs
+1
View File
@@ -79,6 +79,7 @@ pub trait MixnetMessageSender {
// in the 12 bytes format
let context = Context::current();
let trace_id = context.span().span_context().trace_id();
tracing::warn!("Extracted trace_id from surb context: {:?}", trace_id);
let trace_id = compress_trace_id(&trace_id);
let lane = TransmissionLane::General;