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

Compiles but runtime time fails

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2022-10-03 14:00:47 +01:00
parent 4648967e93
commit f4ea183c1e
42 changed files with 3013 additions and 436 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
-2
View File
@@ -663,10 +663,8 @@ dependencies = [
"serde",
"sled",
"tap",
"task",
"tempfile",
"thiserror",
"tokio",
"topology",
"url",
"validator-client",
clients/client-core/Cargo.toml
+15 -6
View File
@@ -13,26 +13,35 @@ humantime-serde = "1.0"
log = "0.4"
rand = { version = "0.7.3", features = ["wasm-bindgen"] }
serde = { version = "1.0", features = ["derive"] }
sled = "0.34"
sled = { version = "0.34", optional = true }
thiserror = "1.0.34"
tokio = { version = "1.19.1", features = ["macros"] }
url = { version ="2.2", features = ["serde"] }
# internal
config = { path = "../../common/config" }
crypto = { path = "../../common/crypto" }
gateway-client = { path = "../../common/client-libs/gateway-client" }
gateway-client = { path = "../../common/client-libs/gateway-client", default-features = false, features = ["wasm", "coconut"] }
gateway-requests = { path = "../../gateway/gateway-requests" }
nonexhaustive-delayqueue = { path = "../../common/nonexhaustive-delayqueue" }
nymsphinx = { path = "../../common/nymsphinx" }
pemstore = { path = "../../common/pemstore" }
task = { path = "../../common/task" }
#task = { path = "../../common/task" }
topology = { path = "../../common/topology" }
validator-client = { path = "../../common/client-libs/validator-client" }
validator-client = { path = "../../common/client-libs/validator-client", default-features = false }
tap = "1.0.1"
tokio = { version = "1.21.2", features = ["time", "macros"]}
[target."cfg(target_arch = \"wasm32\")".dependencies.wasm-bindgen-futures]
version = "0.4"
[dev-dependencies]
tempfile = "3.1.0"
[features]
coconut = ["gateway-client/coconut", "gateway-requests/coconut"]
#wasm = ["gateway-client/wasm"]
default = ["reply-surb"]
wasm = []
#coconut = ["gateway-client/coconut", "gateway-requests/coconut"]
reply-surb = ["sled"]
clients/client-core/src/client/cover_traffic_stream.rs
+36 -25
View File
@@ -3,6 +3,7 @@
use crate::client::mix_traffic::BatchMixMessageSender;
use crate::client::topology_control::TopologyAccessor;
use crate::spawn_future;
use futures::task::{Context, Poll};
use futures::{Future, Stream, StreamExt};
use log::*;
@@ -13,10 +14,11 @@ use nymsphinx::utils::sample_poisson_duration;
use rand::{rngs::OsRng, CryptoRng, Rng};
use std::pin::Pin;
use std::sync::Arc;
use task::ShutdownListener;
use tokio::task::JoinHandle;
use tokio::time;
#[cfg(not(target_arch = "wasm32"))]
use task::ShutdownListener;
pub struct LoopCoverTrafficStream<R>
where
R: CryptoRng + Rng,
@@ -51,6 +53,7 @@ where
topology_access: TopologyAccessor,
/// Listen to shutdown signals.
#[cfg(not(target_arch = "wasm32"))]
shutdown: ShutdownListener,
}
@@ -97,7 +100,7 @@ impl LoopCoverTrafficStream<OsRng> {
mix_tx: BatchMixMessageSender,
our_full_destination: Recipient,
topology_access: TopologyAccessor,
shutdown: ShutdownListener,
#[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener,
) -> Self {
let rng = OsRng;
@@ -111,6 +114,7 @@ impl LoopCoverTrafficStream<OsRng> {
our_full_destination,
rng,
topology_access,
#[cfg(not(target_arch = "wasm32"))]
shutdown,
}
}
@@ -156,6 +160,9 @@ impl LoopCoverTrafficStream<OsRng> {
// JS: due to identical logical structure to OutQueueControl::on_message(), this is also
// presumably required to prevent bugs in the future. Exact reason is still unknown to me.
// TODO: temporary and BAD workaround for wasm (we should find a way to yield here in wasm)
#[cfg(not(target_arch = "wasm32"))]
tokio::task::yield_now().await;
}
@@ -166,30 +173,34 @@ impl LoopCoverTrafficStream<OsRng> {
self.average_cover_message_sending_delay,
)));
let mut shutdown = self.shutdown.clone();
while !shutdown.is_shutdown() {
tokio::select! {
biased;
_ = shutdown.recv() => {
log::trace!("LoopCoverTrafficStream: Received shutdown");
}
next = self.next() => {
if next.is_some() {
self.on_new_message().await;
} else {
log::trace!("LoopCoverTrafficStream: Stopping since channel closed");
break;
}
}
}
// TODO: fix it for non-wasm
while self.next().await.is_some() {
self.on_new_message().await;
}
assert!(self.shutdown.is_shutdown_poll());
log::debug!("LoopCoverTrafficStream: Exiting");
// let mut shutdown = self.shutdown.clone();
// while !shutdown.is_shutdown() {
// tokio::select! {
// biased;
// _ = shutdown.recv() => {
// log::trace!("LoopCoverTrafficStream: Received shutdown");
// }
// next = self.next() => {
// if next.is_some() {
// self.on_new_message().await;
// } else {
// log::trace!("LoopCoverTrafficStream: Stopping since channel closed");
// break;
// }
// }
// }
// }
// assert!(self.shutdown.is_shutdown_poll());
// log::debug!("LoopCoverTrafficStream: Exiting");
}
pub fn start(mut self) -> JoinHandle<()> {
tokio::spawn(async move {
self.run().await;
})
pub fn start(mut self) {
spawn_future(async move { self.run().await })
}
}
clients/client-core/src/client/mix_traffic.rs
+30 -24
View File
@@ -1,13 +1,14 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::spawn_future;
use futures::channel::mpsc;
use futures::StreamExt;
use gateway_client::GatewayClient;
use log::*;
use nymsphinx::forwarding::packet::MixPacket;
#[cfg(not(target_arch = "wasm32"))]
use task::ShutdownListener;
use tokio::task::JoinHandle;
pub type BatchMixMessageSender = mpsc::UnboundedSender<Vec<MixPacket>>;
pub type BatchMixMessageReceiver = mpsc::UnboundedReceiver<Vec<MixPacket>>;
@@ -19,6 +20,7 @@ pub struct MixTrafficController {
// later on gateway_client will need to be accessible by other entities
gateway_client: GatewayClient,
mix_rx: BatchMixMessageReceiver,
#[cfg(not(target_arch = "wasm32"))]
shutdown: ShutdownListener,
// TODO: this is temporary work-around.
@@ -30,13 +32,14 @@ impl MixTrafficController {
pub fn new(
mix_rx: BatchMixMessageReceiver,
gateway_client: GatewayClient,
shutdown: ShutdownListener,
#[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener,
) -> MixTrafficController {
MixTrafficController {
gateway_client,
mix_rx,
shutdown,
consecutive_gateway_failure_count: 0,
#[cfg(not(target_arch = "wasm32"))]
shutdown,
}
}
@@ -70,29 +73,32 @@ impl MixTrafficController {
}
pub async fn run(&mut self) {
while !self.shutdown.is_shutdown() {
tokio::select! {
mix_packets = self.mix_rx.next() => match mix_packets {
Some(mix_packets) => {
self.on_messages(mix_packets).await;
},
None => {
log::trace!("MixTrafficController: Stopping since channel closed");
break;
}
},
_ = self.shutdown.recv() => {
log::trace!("MixTrafficController: Received shutdown");
}
}
// TODO: the usual wasm stuff
while let Some(mix_packets) = self.mix_rx.next().await {
self.on_messages(mix_packets).await;
}
assert!(self.shutdown.is_shutdown_poll());
log::debug!("MixTrafficController: Exiting");
// while !self.shutdown.is_shutdown() {
// tokio::select! {
// mix_packets = self.mix_rx.next() => match mix_packets {
// Some(mix_packets) => {
// self.on_messages(mix_packets).await;
// },
// None => {
// log::trace!("MixTrafficController: Stopping since channel closed");
// break;
// }
// },
// _ = self.shutdown.recv() => {
// log::trace!("MixTrafficController: Received shutdown");
// }
// }
// }
// assert!(self.shutdown.is_shutdown_poll());
// log::debug!("MixTrafficController: Exiting");
}
pub fn start(mut self) -> JoinHandle<()> {
tokio::spawn(async move {
self.run().await;
})
pub fn start(mut self) {
spawn_future(async move { self.run().await })
}
}
clients/client-core/src/client/mod.rs
+1
View File
@@ -6,6 +6,7 @@ pub mod key_manager;
pub mod mix_traffic;
pub mod real_messages_control;
pub mod received_buffer;
#[cfg(feature = "reply-surb")]
pub mod reply_key_storage;
pub mod topology_control;
clients/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs
+35 -20
View File
@@ -10,6 +10,8 @@ use nymsphinx::{
chunking::fragment::{FragmentIdentifier, COVER_FRAG_ID},
};
use std::sync::Arc;
#[cfg(not(target_arch = "wasm32"))]
use task::ShutdownListener;
/// Module responsible for listening for any data resembling acknowledgements from the network
@@ -18,6 +20,7 @@ pub(super) struct AcknowledgementListener {
ack_key: Arc<AckKey>,
ack_receiver: AcknowledgementReceiver,
action_sender: ActionSender,
#[cfg(not(target_arch = "wasm32"))]
shutdown: ShutdownListener,
}
@@ -26,12 +29,13 @@ impl AcknowledgementListener {
ack_key: Arc<AckKey>,
ack_receiver: AcknowledgementReceiver,
action_sender: ActionSender,
shutdown: ShutdownListener,
#[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener,
) -> Self {
AcknowledgementListener {
ack_key,
ack_receiver,
action_sender,
#[cfg(not(target_arch = "wasm32"))]
shutdown,
}
}
@@ -69,26 +73,37 @@ impl AcknowledgementListener {
pub(super) async fn run(&mut self) {
debug!("Started AcknowledgementListener");
while !self.shutdown.is_shutdown() {
tokio::select! {
acks = self.ack_receiver.next() => match acks {
Some(acks) => {
// realistically we would only be getting one ack at the time
for ack in acks {
self.on_ack(ack).await;
}
},
None => {
log::trace!("AcknowledgementListener: Stopping since channel closed");
break;
}
},
_ = self.shutdown.recv() => {
log::trace!("AcknowledgementListener: Received shutdown");
}
// todo: make it non-wasm compatible
debug!("Started AcknowledgementListener");
while let Some(acks) = self.ack_receiver.next().await {
// realistically we would only be getting one ack at the time
for ack in acks {
self.on_ack(ack).await;
}
}
assert!(self.shutdown.is_shutdown_poll());
log::debug!("AcknowledgementListener: Exiting");
error!("TODO: error msg. Or maybe panic?")
// while !self.shutdown.is_shutdown() {
// tokio::select! {
// acks = self.ack_receiver.next() => match acks {
// Some(acks) => {
// // realistically we would only be getting one ack at the time
// for ack in acks {
// self.on_ack(ack).await;
// }
// },
// None => {
// log::trace!("AcknowledgementListener: Stopping since channel closed");
// break;
// }
// },
// _ = self.shutdown.recv() => {
// log::trace!("AcknowledgementListener: Received shutdown");
// }
// }
// }
// assert!(self.shutdown.is_shutdown_poll());
// log::debug!("AcknowledgementListener: Exiting");
}
}
clients/client-core/src/client/real_messages_control/acknowledgement_control/action_controller.rs
+38 -23
View File
@@ -12,6 +12,8 @@ use nymsphinx::Delay as SphinxDelay;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
#[cfg(not(target_arch = "wasm32"))]
use task::ShutdownListener;
pub(crate) type ActionSender = UnboundedSender<Action>;
@@ -102,6 +104,7 @@ pub(super) struct ActionController {
retransmission_sender: RetransmissionRequestSender,
/// Listen for shutdown notifications
#[cfg(not(target_arch = "wasm32"))]
shutdown: ShutdownListener,
}
@@ -109,7 +112,7 @@ impl ActionController {
pub(super) fn new(
config: Config,
retransmission_sender: RetransmissionRequestSender,
shutdown: ShutdownListener,
#[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener,
) -> (Self, ActionSender) {
let (sender, receiver) = mpsc::unbounded();
(
@@ -119,6 +122,7 @@ impl ActionController {
pending_acks_timers: NonExhaustiveDelayQueue::new(),
incoming_actions: receiver,
retransmission_sender,
#[cfg(not(target_arch = "wasm32"))]
shutdown,
},
sender,
@@ -252,30 +256,41 @@ impl ActionController {
}
pub(super) async fn run(&mut self) {
while !self.shutdown.is_shutdown() {
// TODO: shutdown without wasm etc etc
loop {
// at some point there will be a global shutdown signal here as the third option
tokio::select! {
action = self.incoming_actions.next() => match action {
Some(action) => self.process_action(action),
None => {
log::trace!(
"ActionController: Stopping since incoming actions channel closed"
);
break;
}
},
expired_ack = self.pending_acks_timers.next() => match expired_ack {
Some(expired_ack) => self.handle_expired_ack_timer(expired_ack),
None => {
log::trace!("ActionController: Stopping since ack channel closed");
break;
}
},
_ = self.shutdown.recv() => {
log::trace!("ActionController: Received shutdown");
}
// we NEVER expect for ANY sender to get dropped so unwrap here is fine
action = self.incoming_actions.next() => self.process_action(action.unwrap()),
// pending ack queue Stream CANNOT return a `None` so unwrap here is fine
expired_ack = self.pending_acks_timers.next() => self.handle_expired_ack_timer(expired_ack.unwrap())
}
}
assert!(self.shutdown.is_shutdown_poll());
log::debug!("ActionController: Exiting");
// while !self.shutdown.is_shutdown() {
// tokio::select! {
// action = self.incoming_actions.next() => match action {
// Some(action) => self.process_action(action),
// None => {
// log::trace!(
// "ActionController: Stopping since incoming actions channel closed"
// );
// break;
// }
// },
// expired_ack = self.pending_acks_timers.next() => match expired_ack {
// Some(expired_ack) => self.handle_expired_ack_timer(expired_ack),
// None => {
// log::trace!("ActionController: Stopping since ack channel closed");
// break;
// }
// },
// _ = self.shutdown.recv() => {
// log::trace!("ActionController: Received shutdown");
// }
// }
// }
// assert!(self.shutdown.is_shutdown_poll());
// log::debug!("ActionController: Exiting");
}
}
clients/client-core/src/client/real_messages_control/acknowledgement_control/input_message_listener.rs
+35 -20
View File
@@ -3,7 +3,6 @@
use super::action_controller::{Action, ActionSender};
use super::PendingAcknowledgement;
use crate::client::reply_key_storage::ReplyKeyStorage;
use crate::client::{
inbound_messages::{InputMessage, InputMessageReceiver},
real_messages_control::real_traffic_stream::{BatchRealMessageSender, RealMessage},
@@ -16,6 +15,11 @@ use nymsphinx::preparer::MessagePreparer;
use nymsphinx::{acknowledgements::AckKey, addressing::clients::Recipient};
use rand::{CryptoRng, Rng};
use std::sync::Arc;
#[cfg(feature = "reply-surb")]
use crate::client::reply_key_storage::ReplyKeyStorage;
#[cfg(not(target_arch = "wasm32"))]
use task::ShutdownListener;
/// Module responsible for dealing with the received messages: splitting them, creating acknowledgements,
@@ -32,7 +36,9 @@ where
action_sender: ActionSender,
real_message_sender: BatchRealMessageSender,
topology_access: TopologyAccessor,
#[cfg(feature = "reply-surb")]
reply_key_storage: ReplyKeyStorage,
#[cfg(not(target_arch = "wasm32"))]
shutdown: ShutdownListener,
}
@@ -51,8 +57,8 @@ where
action_sender: ActionSender,
real_message_sender: BatchRealMessageSender,
topology_access: TopologyAccessor,
reply_key_storage: ReplyKeyStorage,
shutdown: ShutdownListener,
#[cfg(feature = "reply-surb")] reply_key_storage: ReplyKeyStorage,
#[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener,
) -> Self {
InputMessageListener {
ack_key,
@@ -62,7 +68,9 @@ where
action_sender,
real_message_sender,
topology_access,
#[cfg(feature = "reply-surb")]
reply_key_storage,
#[cfg(not(target_arch = "wasm32"))]
shutdown,
}
}
@@ -121,6 +129,7 @@ where
.prepare_and_split_message(content, with_reply_surb, topology)
.expect("somehow the topology was invalid after all!");
#[cfg(feature = "reply-surb")]
if let Some(reply_key) = reply_key {
self.reply_key_storage
.insert_encryption_key(reply_key)
@@ -185,24 +194,30 @@ where
}
pub(super) async fn run(&mut self) {
// TODO: shutdown without wasm etc etc
debug!("Started InputMessageListener");
while !self.shutdown.is_shutdown() {
tokio::select! {
input_msg = self.input_receiver.next() => match input_msg {
Some(input_msg) => {
self.on_input_message(input_msg).await;
},
None => {
log::trace!("InputMessageListener: Stopping since channel closed");
break;
}
},
_ = self.shutdown.recv() => {
log::trace!("InputMessageListener: Received shutdown");
}
}
while let Some(input_msg) = self.input_receiver.next().await {
self.on_input_message(input_msg).await;
}
assert!(self.shutdown.is_shutdown_poll());
log::debug!("InputMessageListener: Exiting");
// debug!("Started InputMessageListener");
// while !self.shutdown.is_shutdown() {
// tokio::select! {
// input_msg = self.input_receiver.next() => match input_msg {
// Some(input_msg) => {
// self.on_input_message(input_msg).await;
// },
// None => {
// log::trace!("InputMessageListener: Stopping since channel closed");
// break;
// }
// },
// _ = self.shutdown.recv() => {
// log::trace!("InputMessageListener: Received shutdown");
// }
// }
// }
// assert!(self.shutdown.is_shutdown_poll());
// log::debug!("InputMessageListener: Exiting");
}
}
clients/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs
+39 -31
View File
@@ -8,8 +8,8 @@ use self::{
sent_notification_listener::SentNotificationListener,
};
use super::real_traffic_stream::BatchRealMessageSender;
use crate::client::reply_key_storage::ReplyKeyStorage;
use crate::client::{inbound_messages::InputMessageReceiver, topology_control::TopologyAccessor};
use crate::spawn_future;
use futures::channel::mpsc;
use gateway_client::AcknowledgementReceiver;
use log::*;
@@ -25,8 +25,12 @@ use std::{
sync::{Arc, Weak},
time::Duration,
};
#[cfg(feature = "reply-surb")]
use crate::client::reply_key_storage::ReplyKeyStorage;
#[cfg(not(target_arch = "wasm32"))]
use task::ShutdownListener;
use tokio::task::JoinHandle;
mod acknowledgement_listener;
mod action_controller;
@@ -160,16 +164,20 @@ where
topology_access: TopologyAccessor,
ack_key: Arc<AckKey>,
ack_recipient: Recipient,
reply_key_storage: ReplyKeyStorage,
connectors: AcknowledgementControllerConnectors,
shutdown: ShutdownListener,
#[cfg(feature = "reply-surb")] reply_key_storage: ReplyKeyStorage,
#[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener,
) -> Self {
let (retransmission_tx, retransmission_rx) = mpsc::unbounded();
let action_config =
action_controller::Config::new(config.ack_wait_addition, config.ack_wait_multiplier);
let (action_controller, action_sender) =
ActionController::new(action_config, retransmission_tx, shutdown.clone());
let (action_controller, action_sender) = ActionController::new(
action_config,
retransmission_tx,
#[cfg(not(target_arch = "wasm32"))]
shutdown.clone(),
);
let message_preparer = MessagePreparer::new(
rng,
@@ -183,6 +191,7 @@ where
Arc::clone(&ack_key),
connectors.ack_receiver,
action_sender.clone(),
#[cfg(not(target_arch = "wasm32"))]
shutdown.clone(),
);
@@ -195,7 +204,9 @@ where
action_sender.clone(),
connectors.real_message_sender.clone(),
topology_access.clone(),
#[cfg(feature = "reply-surb")]
reply_key_storage,
#[cfg(not(target_arch = "wasm32"))]
shutdown.clone(),
);
@@ -208,13 +219,18 @@ where
connectors.real_message_sender,
retransmission_rx,
topology_access,
#[cfg(not(target_arch = "wasm32"))]
shutdown.clone(),
);
// will listen for events indicating the packet was sent through the network so that
// the retransmission timer should be started.
let sent_notification_listener =
SentNotificationListener::new(connectors.sent_notifier, action_sender, shutdown);
let sent_notification_listener = SentNotificationListener::new(
connectors.sent_notifier,
action_sender,
#[cfg(not(target_arch = "wasm32"))]
shutdown,
);
AcknowledgementController {
acknowledgement_listener: Some(acknowledgement_listener),
@@ -236,47 +252,39 @@ where
// the below are log messages are errors as at the current stage we do not expect any of
// the task to ever finish. This will of course change once we introduce
// graceful shutdowns.
let ack_listener_fut = tokio::spawn(async move {
let ack_listener_fut = spawn_future(async move {
acknowledgement_listener.run().await;
debug!("The acknowledgement listener has finished execution!");
acknowledgement_listener
});
let input_listener_fut = tokio::spawn(async move {
let input_listener_fut = spawn_future(async move {
input_message_listener.run().await;
debug!("The input listener has finished execution!");
input_message_listener
});
let retransmission_req_fut = tokio::spawn(async move {
let retransmission_req_fut = spawn_future(async move {
retransmission_request_listener.run().await;
debug!("The retransmission request listener has finished execution!");
retransmission_request_listener
});
let sent_notification_fut = tokio::spawn(async move {
let sent_notification_fut = spawn_future(async move {
sent_notification_listener.run().await;
debug!("The sent notification listener has finished execution!");
sent_notification_listener
});
let action_controller_fut = tokio::spawn(async move {
let action_controller_fut = spawn_future(async move {
action_controller.run().await;
debug!("The controller has finished execution!");
action_controller
});
// technically we don't have to bring `AcknowledgementController` back to a valid state
// but we can do it, so why not? Perhaps it might be useful if we wanted to allow
// for restarts of certain modules without killing the entire process.
self.acknowledgement_listener = Some(ack_listener_fut.await.unwrap());
self.input_message_listener = Some(input_listener_fut.await.unwrap());
self.retransmission_request_listener = Some(retransmission_req_fut.await.unwrap());
self.sent_notification_listener = Some(sent_notification_fut.await.unwrap());
self.action_controller = Some(action_controller_fut.await.unwrap());
// // technically we don't have to bring `AcknowledgementController` back to a valid state
// // but we can do it, so why not? Perhaps it might be useful if we wanted to allow
// // for restarts of certain modules without killing the entire process.
// self.acknowledgement_listener = Some(ack_listener_fut.await.unwrap());
// self.input_message_listener = Some(input_listener_fut.await.unwrap());
// self.retransmission_request_listener = Some(retransmission_req_fut.await.unwrap());
// self.sent_notification_listener = Some(sent_notification_fut.await.unwrap());
// self.action_controller = Some(action_controller_fut.await.unwrap());
}
#[allow(dead_code)]
pub(super) fn start(mut self) -> JoinHandle<Self> {
tokio::spawn(async move {
self.run().await;
self
})
pub(super) fn start(mut self) {
spawn_future(async move { self.run().await })
}
}
clients/client-core/src/client/real_messages_control/acknowledgement_control/retransmission_request_listener.rs
+25 -16
View File
@@ -14,6 +14,8 @@ use nymsphinx::preparer::MessagePreparer;
use nymsphinx::{acknowledgements::AckKey, addressing::clients::Recipient};
use rand::{CryptoRng, Rng};
use std::sync::{Arc, Weak};
#[cfg(not(target_arch = "wasm32"))]
use task::ShutdownListener;
// responsible for packet retransmission upon fired timer
@@ -28,6 +30,7 @@ where
real_message_sender: BatchRealMessageSender,
request_receiver: RetransmissionRequestReceiver,
topology_access: TopologyAccessor,
#[cfg(not(target_arch = "wasm32"))]
shutdown: ShutdownListener,
}
@@ -44,7 +47,7 @@ where
real_message_sender: BatchRealMessageSender,
request_receiver: RetransmissionRequestReceiver,
topology_access: TopologyAccessor,
shutdown: ShutdownListener,
#[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener,
) -> Self {
RetransmissionRequestListener {
ack_key,
@@ -54,6 +57,7 @@ where
real_message_sender,
request_receiver,
topology_access,
#[cfg(not(target_arch = "wasm32"))]
shutdown,
}
}
@@ -127,21 +131,26 @@ where
pub(super) async fn run(&mut self) {
debug!("Started RetransmissionRequestListener");
while !self.shutdown.is_shutdown() {
tokio::select! {
timed_out_ack = self.request_receiver.next() => match timed_out_ack {
Some(timed_out_ack) => self.on_retransmission_request(timed_out_ack).await,
None => {
log::trace!("RetransmissionRequestListener: Stopping since channel closed");
break;
}
},
_ = self.shutdown.recv() => {
log::trace!("RetransmissionRequestListener: Received shutdown");
}
}
// TODO: shutdown without wasm etc etc
while let Some(timed_out_ack) = self.request_receiver.next().await {
self.on_retransmission_request(timed_out_ack).await;
}
assert!(self.shutdown.is_shutdown_poll());
log::debug!("RetransmissionRequestListener: Exiting");
// while !self.shutdown.is_shutdown() {
// tokio::select! {
// timed_out_ack = self.request_receiver.next() => match timed_out_ack {
// Some(timed_out_ack) => self.on_retransmission_request(timed_out_ack).await,
// None => {
// log::trace!("RetransmissionRequestListener: Stopping since channel closed");
// break;
// }
// },
// _ = self.shutdown.recv() => {
// log::trace!("RetransmissionRequestListener: Received shutdown");
// }
// }
// }
// assert!(self.shutdown.is_shutdown_poll());
// log::debug!("RetransmissionRequestListener: Exiting");
}
}
clients/client-core/src/client/real_messages_control/acknowledgement_control/sent_notification_listener.rs
+28 -18
View File
@@ -6,6 +6,8 @@ use super::SentPacketNotificationReceiver;
use futures::StreamExt;
use log::*;
use nymsphinx::chunking::fragment::{FragmentIdentifier, COVER_FRAG_ID};
#[cfg(not(target_arch = "wasm32"))]
use task::ShutdownListener;
/// Module responsible for starting up retransmission timers.
@@ -15,6 +17,7 @@ use task::ShutdownListener;
pub(super) struct SentNotificationListener {
sent_notifier: SentPacketNotificationReceiver,
action_sender: ActionSender,
#[cfg(not(target_arch = "wasm32"))]
shutdown: ShutdownListener,
}
@@ -22,11 +25,12 @@ impl SentNotificationListener {
pub(super) fn new(
sent_notifier: SentPacketNotificationReceiver,
action_sender: ActionSender,
shutdown: ShutdownListener,
#[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener,
) -> Self {
SentNotificationListener {
sent_notifier,
action_sender,
#[cfg(not(target_arch = "wasm32"))]
shutdown,
}
}
@@ -48,23 +52,29 @@ impl SentNotificationListener {
pub(super) async fn run(&mut self) {
debug!("Started SentNotificationListener");
while !self.shutdown.is_shutdown() {
tokio::select! {
frag_id = self.sent_notifier.next() => match frag_id {
Some(frag_id) => {
self.on_sent_message(frag_id).await;
}
None => {
log::trace!("SentNotificationListener: Stopping since channel closed");
break;
}
},
_ = self.shutdown.recv() => {
log::trace!("SentNotificationListener: Received shutdown");
}
}
// TODO: shutdown without wasm etc etc
while let Some(frag_id) = self.sent_notifier.next().await {
self.on_sent_message(frag_id).await;
}
assert!(self.shutdown.is_shutdown_poll());
log::debug!("SentNotificationListener: Exiting");
// while !self.shutdown.is_shutdown() {
// tokio::select! {
// frag_id = self.sent_notifier.next() => match frag_id {
// Some(frag_id) => {
// self.on_sent_message(frag_id).await;
// }
// None => {
// log::trace!("SentNotificationListener: Stopping since channel closed");
// break;
// }
// },
// _ = self.shutdown.recv() => {
// log::trace!("SentNotificationListener: Received shutdown");
// }
// }
// }
// assert!(self.shutdown.is_shutdown_poll());
// log::debug!("SentNotificationListener: Exiting");
}
}
clients/client-core/src/client/real_messages_control/mod.rs
+21 -19
View File
@@ -9,11 +9,11 @@ use self::{
acknowledgement_control::AcknowledgementController, real_traffic_stream::OutQueueControl,
};
use crate::client::real_messages_control::acknowledgement_control::AcknowledgementControllerConnectors;
use crate::client::reply_key_storage::ReplyKeyStorage;
use crate::client::{
inbound_messages::InputMessageReceiver, mix_traffic::BatchMixMessageSender,
topology_control::TopologyAccessor,
};
use crate::spawn_future;
use futures::channel::mpsc;
use gateway_client::AcknowledgementReceiver;
use log::*;
@@ -22,8 +22,12 @@ use nymsphinx::addressing::clients::Recipient;
use rand::{rngs::OsRng, CryptoRng, Rng};
use std::sync::Arc;
use std::time::Duration;
#[cfg(feature = "reply-surb")]
use crate::client::reply_key_storage::ReplyKeyStorage;
#[cfg(not(target_arch = "wasm32"))]
use task::ShutdownListener;
use tokio::task::JoinHandle;
mod acknowledgement_control;
mod real_traffic_stream;
@@ -91,8 +95,8 @@ impl RealMessagesController<OsRng> {
input_receiver: InputMessageReceiver,
mix_sender: BatchMixMessageSender,
topology_access: TopologyAccessor,
reply_key_storage: ReplyKeyStorage,
shutdown: ShutdownListener,
#[cfg(feature = "reply-surb")] reply_key_storage: ReplyKeyStorage,
#[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener,
) -> Self {
let rng = OsRng;
@@ -119,8 +123,10 @@ impl RealMessagesController<OsRng> {
topology_access.clone(),
Arc::clone(&config.ack_key),
config.self_recipient,
reply_key_storage,
ack_controller_connectors,
#[cfg(feature = "reply-surb")]
reply_key_storage,
#[cfg(not(target_arch = "wasm32"))]
shutdown.clone(),
);
@@ -139,6 +145,7 @@ impl RealMessagesController<OsRng> {
rng,
config.self_recipient,
topology_access,
#[cfg(not(target_arch = "wasm32"))]
shutdown,
);
@@ -155,28 +162,23 @@ impl RealMessagesController<OsRng> {
// the below are log messages are errors as at the current stage we do not expect any of
// the task to ever finish. This will of course change once we introduce
// graceful shutdowns.
let out_queue_control_fut = tokio::spawn(async move {
let out_queue_control_fut = spawn_future(async move {
out_queue_control.run_out_queue_control().await;
debug!("The out queue controller has finished execution!");
out_queue_control
});
let ack_control_fut = tokio::spawn(async move {
let ack_control_fut = spawn_future(async move {
ack_control.run().await;
debug!("The acknowledgement controller has finished execution!");
ack_control
});
// technically we don't have to bring `RealMessagesController` back to a valid state
// but we can do it, so why not? Perhaps it might be useful if we wanted to allow
// for restarts of certain modules without killing the entire process.
self.out_queue_control = Some(out_queue_control_fut.await.unwrap());
self.ack_control = Some(ack_control_fut.await.unwrap());
// // technically we don't have to bring `RealMessagesController` back to a valid state
// // but we can do it, so why not? Perhaps it might be useful if we wanted to allow
// // for restarts of certain modules without killing the entire process.
// self.out_queue_control = Some(out_queue_control_fut.await.unwrap());
// self.ack_control = Some(ack_control_fut.await.unwrap());
}
pub fn start(mut self) -> JoinHandle<Self> {
tokio::spawn(async move {
self.run().await;
self
})
pub fn start(mut self) {
spawn_future(async move { self.run().await })
}
}
clients/client-core/src/client/real_messages_control/real_traffic_stream.rs
+35 -21
View File
@@ -19,9 +19,11 @@ use std::collections::VecDeque;
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;
use task::ShutdownListener;
use tokio::time;
#[cfg(not(target_arch = "wasm32"))]
use task::ShutdownListener;
/// Configurable parameters of the `OutQueueControl`
pub(crate) struct Config {
/// Average delay an acknowledgement packet is going to get delay at a single mixnode.
@@ -86,6 +88,7 @@ where
received_buffer: VecDeque<RealMessage>,
/// Listens for shutdown signals
#[cfg(not(target_arch = "wasm32"))]
shutdown: ShutdownListener,
}
@@ -178,7 +181,7 @@ where
rng: R,
our_full_destination: Recipient,
topology_access: TopologyAccessor,
shutdown: ShutdownListener,
#[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener,
) -> Self {
OutQueueControl {
config,
@@ -191,6 +194,7 @@ where
rng,
topology_access,
received_buffer: VecDeque::with_capacity(0), // we won't be putting any data into this guy directly
#[cfg(not(target_arch = "wasm32"))]
shutdown,
}
}
@@ -246,6 +250,7 @@ where
// - the receiver channel is closed
// in either case there's no recovery and we can only panic
if let Err(err) = self.mix_tx.unbounded_send(vec![next_message]) {
#[cfg(not(target_arch = "wasm32"))]
if self.shutdown.is_shutdown_poll() {
log::info!("Failed to send (shutdown detected)");
} else {
@@ -260,6 +265,9 @@ where
// JS2: Basically it was the case that with high enough rate, the stream had already a next value
// ready and hence was immediately re-scheduled causing other tasks to be starved;
// yield makes it go back the scheduling queue regardless of its value availability
// TODO: temporary and BAD workaround for wasm (we should find a way to yield here in wasm)
#[cfg(not(target_arch = "wasm32"))]
tokio::task::yield_now().await;
}
@@ -271,26 +279,32 @@ where
self.config.average_message_sending_delay,
)));
let mut shutdown = self.shutdown.clone();
while !shutdown.is_shutdown() {
tokio::select! {
biased;
_ = shutdown.recv() => {
log::trace!("OutQueueControl: Received shutdown");
}
next_message = self.next() => match next_message {
Some(next_message) => {
self.on_message(next_message).await;
},
None => {
log::trace!("OutQueueControl: Stopping since channel closed");
break;
}
}
}
// TODO: fix it for non-wasm
while let Some(next_message) = self.next().await {
self.on_message(next_message).await;
}
assert!(shutdown.is_shutdown_poll());
log::debug!("OutQueueControl: Exiting");
// let mut shutdown = self.shutdown.clone();
// while !shutdown.is_shutdown() {
// tokio::select! {
// biased;
// _ = shutdown.recv() => {
// log::trace!("OutQueueControl: Received shutdown");
// }
// next_message = self.next() => match next_message {
// Some(next_message) => {
// self.on_message(next_message).await;
// },
// None => {
// log::trace!("OutQueueControl: Stopping since channel closed");
// break;
// }
// }
// }
// }
// assert!(shutdown.is_shutdown_poll());
// log::debug!("OutQueueControl: Exiting");
}
pub(crate) async fn run_out_queue_control(&mut self) {
clients/client-core/src/client/received_buffer.rs
+54 -29
View File
@@ -1,8 +1,8 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::client::reply_key_storage::ReplyKeyStorage;
use crate::client::SHUTDOWN_HAS_BEEN_SIGNALLED;
use crate::spawn_future;
use crypto::asymmetric::encryption;
use crypto::symmetric::stream_cipher;
use crypto::Digest;
@@ -17,8 +17,11 @@ use nymsphinx::receiver::{MessageReceiver, MessageRecoveryError, ReconstructedMe
use std::collections::HashSet;
use std::sync::atomic::Ordering;
use std::sync::Arc;
#[cfg(not(target_arch = "wasm32"))]
use crate::client::reply_key_storage::ReplyKeyStorage;
#[cfg(not(target_arch = "wasm32"))]
use task::ShutdownListener;
use tokio::task::JoinHandle;
// Buffer Requests to say "hey, send any reconstructed messages to this channel"
// or to say "hey, I'm going offline, don't send anything more to me. Just buffer them instead"
@@ -116,13 +119,14 @@ struct ReceivedMessagesBuffer {
/// Storage containing keys to all [`ReplySURB`]s ever sent out that we did not receive back.
// There's no need to put it behind a Mutex since it's already properly concurrent
#[cfg(feature = "reply-surb")]
reply_key_storage: ReplyKeyStorage,
}
impl ReceivedMessagesBuffer {
fn new(
local_encryption_keypair: Arc<encryption::KeyPair>,
reply_key_storage: ReplyKeyStorage,
#[cfg(feature = "reply-surb")] reply_key_storage: ReplyKeyStorage,
) -> Self {
ReceivedMessagesBuffer {
inner: Arc::new(Mutex::new(ReceivedMessagesBufferInner {
@@ -132,6 +136,7 @@ impl ReceivedMessagesBuffer {
message_sender: None,
recently_reconstructed: HashSet::new(),
})),
#[cfg(feature = "reply-surb")]
reply_key_storage,
}
}
@@ -180,6 +185,7 @@ impl ReceivedMessagesBuffer {
self.inner.lock().await.messages.extend(msgs)
}
#[cfg(feature = "reply-surb")]
fn process_received_reply(
reply_ciphertext: &[u8],
reply_key: SurbEncryptionKey,
@@ -227,6 +233,7 @@ impl ReceivedMessagesBuffer {
// TODO: this might be a bottleneck - since the keys are stored on disk we, presumably,
// are doing a disk operation every single received fragment
#[cfg(feature = "reply-surb")]
if let Some(reply_encryption_key) = self
.reply_key_storage
.get_and_remove_encryption_key(possible_key_digest)
@@ -244,6 +251,13 @@ impl ReceivedMessagesBuffer {
completed_messages.push(completed_message)
}
}
// TODO: make it nicer and stuff
#[cfg(not(feature = "reply-surb"))]
if let Some(completed_message) = inner_guard.process_received_fragment(msg) {
completed_messages.push(completed_message)
}
}
if !completed_messages.is_empty() {
@@ -293,8 +307,8 @@ impl RequestReceiver {
}
}
fn start(mut self) -> JoinHandle<()> {
tokio::spawn(async move {
fn start(mut self) {
spawn_future(async move {
loop {
tokio::select! {
request = self.query_receiver.next() => {
@@ -323,6 +337,7 @@ impl RequestReceiver {
struct FragmentedMessageReceiver {
received_buffer: ReceivedMessagesBuffer,
mixnet_packet_receiver: MixnetMessageReceiver,
#[cfg(not(target_arch = "wasm32"))]
shutdown: ShutdownListener,
}
@@ -330,34 +345,40 @@ impl FragmentedMessageReceiver {
fn new(
received_buffer: ReceivedMessagesBuffer,
mixnet_packet_receiver: MixnetMessageReceiver,
shutdown: ShutdownListener,
#[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener,
) -> Self {
FragmentedMessageReceiver {
received_buffer,
mixnet_packet_receiver,
#[cfg(not(target_arch = "wasm32"))]
shutdown,
}
}
fn start(mut self) -> JoinHandle<()> {
tokio::spawn(async move {
while !self.shutdown.is_shutdown() {
tokio::select! {
new_messages = self.mixnet_packet_receiver.next() => match new_messages {
Some(new_messages) => {
self.received_buffer.handle_new_received(new_messages).await;
}
None => {
log::trace!("FragmentedMessageReceiver: Stopping since channel closed");
break;
}
},
_ = self.shutdown.recv() => {
log::trace!("FragmentedMessageReceiver: Received shutdown");
}
}
fn start(mut self) {
// TODO: wasm and stuff
spawn_future(async move {
while let Some(new_messages) = self.mixnet_packet_receiver.next().await {
self.received_buffer.handle_new_received(new_messages).await;
}
assert!(self.shutdown.is_shutdown_poll());
log::debug!("FragmentedMessageReceiver: Exiting");
// while !self.shutdown.is_shutdown() {
// tokio::select! {
// new_messages = self.mixnet_packet_receiver.next() => match new_messages {
// Some(new_messages) => {
// self.received_buffer.handle_new_received(new_messages).await;
// }
// None => {
// log::trace!("FragmentedMessageReceiver: Stopping since channel closed");
// break;
// }
// },
// _ = self.shutdown.recv() => {
// log::trace!("FragmentedMessageReceiver: Received shutdown");
// }
// }
// }
// assert!(self.shutdown.is_shutdown_poll());
// log::debug!("FragmentedMessageReceiver: Exiting");
})
}
}
@@ -372,16 +393,20 @@ impl ReceivedMessagesBufferController {
local_encryption_keypair: Arc<encryption::KeyPair>,
query_receiver: ReceivedBufferRequestReceiver,
mixnet_packet_receiver: MixnetMessageReceiver,
reply_key_storage: ReplyKeyStorage,
shutdown: ShutdownListener,
#[cfg(feature = "reply-surb")] reply_key_storage: ReplyKeyStorage,
#[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener,
) -> Self {
let received_buffer =
ReceivedMessagesBuffer::new(local_encryption_keypair, reply_key_storage);
let received_buffer = ReceivedMessagesBuffer::new(
local_encryption_keypair,
#[cfg(feature = "reply-surb")]
reply_key_storage,
);
ReceivedMessagesBufferController {
fragmented_message_receiver: FragmentedMessageReceiver::new(
received_buffer.clone(),
mixnet_packet_receiver,
#[cfg(not(target_arch = "wasm32"))]
shutdown,
),
request_receiver: RequestReceiver::new(received_buffer, query_receiver),
clients/client-core/src/client/topology_control.rs
+25 -15
View File
@@ -1,6 +1,7 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::spawn_future;
use log::*;
use nymsphinx::addressing::clients::Recipient;
use nymsphinx::params::DEFAULT_NUM_MIX_HOPS;
@@ -10,12 +11,13 @@ use std::ops::Deref;
use std::sync::Arc;
use std::time;
use std::time::Duration;
use task::ShutdownListener;
use tokio::sync::{RwLock, RwLockReadGuard};
use tokio::task::JoinHandle;
use topology::{nym_topology_from_bonds, NymTopology};
use url::Url;
#[cfg(not(target_arch = "wasm32"))]
use task::ShutdownListener;
// I'm extremely curious why compiler NEVER complained about lack of Debug here before
#[derive(Debug)]
pub struct TopologyAccessorInner(Option<NymTopology>);
@@ -304,20 +306,28 @@ impl TopologyRefresher {
self.topology_accessor.is_routable().await
}
pub fn start(mut self, mut shutdown: ShutdownListener) -> JoinHandle<()> {
tokio::spawn(async move {
while !shutdown.is_shutdown() {
tokio::select! {
_ = tokio::time::sleep(self.refresh_rate) => {
self.refresh().await;
},
_ = shutdown.recv() => {
log::trace!("TopologyRefresher: Received shutdown");
},
}
pub fn start(mut self, #[cfg(not(target_arch = "wasm32"))] mut shutdown: ShutdownListener) {
// TODO: usual non-wasm, etc, etc
spawn_future(async move {
loop {
// TODO: this won't work in wasm
tokio::time::sleep(self.refresh_rate).await;
self.refresh().await;
}
assert!(shutdown.is_shutdown_poll());
log::debug!("TopologyRefresher: Exiting");
// while !shutdown.is_shutdown() {
// tokio::select! {
// _ = tokio::time::sleep(self.refresh_rate) => {
// self.refresh().await;
// },
// _ = shutdown.recv() => {
// log::trace!("TopologyRefresher: Received shutdown");
// },
// }
// }
// assert!(shutdown.is_shutdown_poll());
// log::debug!("TopologyRefresher: Exiting");
})
}
}
clients/client-core/src/init.rs
+1
View File
@@ -90,6 +90,7 @@ async fn register_with_gateway(
gateway.owner.clone(),
our_identity.clone(),
timeout,
#[cfg(not(target_arch = "wasm32"))]
None,
);
gateway_client
clients/client-core/src/lib.rs
+14
View File
@@ -1,4 +1,18 @@
use std::future::Future;
pub mod client;
pub mod config;
pub mod error;
pub mod init;
// for now we're losing the output but we never really cared about it anyway
pub(crate) fn spawn_future<F>(future: F)
where
F: Future<Output = ()> + 'static,
{
#[cfg(not(target_arch = "wasm32"))]
tokio::spawn(future);
#[cfg(target_arch = "wasm32")]
wasm_bindgen_futures::spawn_local(future);
}
clients/credential/Cargo.toml
+1 -1
View File
@@ -15,7 +15,7 @@ rand = "0.7.3"
serde = { version = "1.0", features = ["derive"] }
thiserror = "1.0"
url = "2.2"
tokio = { version = "1.19.1", features = ["rt-multi-thread", "net", "signal", "macros"] } # async runtime
tokio = { version = "1.21.2", features = ["rt-multi-thread", "net", "signal", "macros"] } # async runtime
coconut-interface = { path = "../../common/coconut-interface" }
config = { path = "../../common/config" }
clients/native/Cargo.toml
+1 -1
View File
@@ -27,7 +27,7 @@ pretty_env_logger = "0.4" # for formatting log messages
rand = { version = "0.7.3", features = ["wasm-bindgen"] } # rng-related traits + some rng implementation to use
serde = { version = "1.0.104", features = ["derive"] } # for config serialization/deserialization
sled = "0.34" # for storage of replySURB decryption keys
tokio = { version = "1.19.1", features = ["rt-multi-thread", "net", "signal"] } # async runtime
tokio = { version = "1.21.2", features = ["rt-multi-thread", "net", "signal"] } # async runtime
tokio-tungstenite = "0.14" # websocket
## internal
clients/socks5/Cargo.toml
+1 -1
View File
@@ -20,7 +20,7 @@ pretty_env_logger = "0.4"
rand = { version = "0.7.3", features = ["wasm-bindgen"] }
serde = { version = "1.0", features = ["derive"] } # for config serialization/deserialization
snafu = "0.6"
tokio = { version = "1.19.1", features = ["rt-multi-thread", "net", "signal"] }
tokio = { version = "1.21.2", features = ["rt-multi-thread", "net", "signal"] }
url = "2.2"
# internal
clients/webassembly/Cargo.toml
+1
View File
@@ -27,6 +27,7 @@ rand = { version = "0.7.3", features = ["wasm-bindgen"] }
url = "2.2"
# internal
client-core = { path = "../client-core", default-features = false, features = ["wasm"] }
coconut-interface = { path = "../../common/coconut-interface", optional = true }
credentials = { path = "../../common/credentials", optional = true }
crypto = { path = "../../common/crypto" }
clients/webassembly/js-example/index.js
+1 -1
View File
@@ -38,7 +38,7 @@ async function main() {
// this is current limitation of wasm in rust - for async methods you can't take self my reference...
// I'm trying to figure out if I can somehow hack my way around it, but for time being you have to re-assign
// the object (it's the same one)
client = await client.initial_setup();
client = await client.start();
const self_address = client.self_address();
displaySenderAddress(self_address);
clients/webassembly/js-example/yarn.lock
+2175
View File
File diff suppressed because it is too large Load Diff
clients/webassembly/src/client/mod.rs
+366 -114
View File
@@ -1,9 +1,27 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use client_core::client::cover_traffic_stream::LoopCoverTrafficStream;
use client_core::client::inbound_messages::InputMessage;
use client_core::client::inbound_messages::InputMessageReceiver;
use client_core::client::key_manager::KeyManager;
use client_core::client::mix_traffic::BatchMixMessageReceiver;
use client_core::client::mix_traffic::BatchMixMessageSender;
use client_core::client::mix_traffic::MixTrafficController;
use client_core::client::real_messages_control;
use client_core::client::real_messages_control::RealMessagesController;
use client_core::client::received_buffer::ReceivedBufferRequestReceiver;
use client_core::client::received_buffer::ReceivedMessagesBufferController;
use client_core::client::topology_control::TopologyAccessor;
use client_core::client::topology_control::TopologyRefresher;
use client_core::client::topology_control::TopologyRefresherConfig;
use crypto::asymmetric::{encryption, identity};
use futures::channel::mpsc;
use gateway_client::AcknowledgementReceiver;
use gateway_client::AcknowledgementSender;
use gateway_client::GatewayClient;
use gateway_client::MixnetMessageReceiver;
use gateway_client::MixnetMessageSender;
use nymsphinx::acknowledgements::AckKey;
use nymsphinx::addressing::clients::Recipient;
use nymsphinx::preparer::MessagePreparer;
@@ -19,29 +37,36 @@ use wasm_utils::{console_log, console_warn};
pub(crate) mod received_processor;
const DEFAULT_AVERAGE_PACKET_DELAY: Duration = Duration::from_millis(200);
const DEFAULT_AVERAGE_ACK_DELAY: Duration = Duration::from_millis(200);
const DEFAULT_GATEWAY_RESPONSE_TIMEOUT: Duration = Duration::from_millis(1_500);
// TODO: make those properly configurable later
const ACK_WAIT_MULTIPLIER: f64 = 1.5;
const ACK_WAIT_ADDITION: Duration = Duration::from_millis(1_500);
const LOOP_COVER_STREAM_AVERAGE_DELAY: Duration = Duration::from_millis(200);
const MESSAGE_STREAM_AVERAGE_DELAY: Duration = Duration::from_millis(20);
const AVERAGE_PACKET_DELAY: Duration = Duration::from_millis(50);
const AVERAGE_ACK_DELAY: Duration = Duration::from_millis(50);
const TOPOLOGY_REFRESH_RATE: Duration = Duration::from_secs(5 * 60);
const TOPOLOGY_RESOLUTION_TIMEOUT: Duration = Duration::from_millis(5_000);
const GATEWAY_RESPONSE_TIMEOUT: Duration = Duration::from_millis(1_500);
#[wasm_bindgen]
pub struct NymClient {
validator_server: Url,
disabled_credentials_mode: bool,
// TODO: technically this doesn't need to be an Arc since wasm is run on a single thread
// however, once we eventually combine this code with the native-client's, it will make things
// easier.
identity: Arc<identity::KeyPair>,
encryption_keys: Arc<encryption::KeyPair>,
ack_key: Arc<AckKey>,
/// KeyManager object containing smart pointers to all relevant keys used by the client.
key_manager: KeyManager,
message_preparer: Option<MessagePreparer<OsRng>>,
// message_preparer: Option<MessagePreparer<OsRng>>,
// message_receiver: MessageReceiver,
// TODO: this should be stored somewhere persistently
// received_keys: HashSet<SURBEncryptionKey>,
// TODO: only temporary
topology: Option<NymTopology>,
gateway_client: Option<GatewayClient>,
// gateway_client: Option<GatewayClient>,
gateway_identity: Option<identity::PublicKey>,
// callbacks
on_message: Option<js_sys::Function>,
@@ -54,22 +79,19 @@ impl NymClient {
pub fn new(validator_server: String) -> Self {
let mut rng = OsRng;
// for time being generate new keys each time...
let identity = identity::KeyPair::new(&mut rng);
let encryption_keys = encryption::KeyPair::new(&mut rng);
let ack_key = AckKey::new(&mut rng);
let mut key_manager = KeyManager::new(&mut rng);
console_log!("generated new set of keys");
Self {
identity: Arc::new(identity),
encryption_keys: Arc::new(encryption_keys),
ack_key: Arc::new(ack_key),
key_manager,
validator_server: validator_server
.parse()
.expect("malformed validator server url provided"),
message_preparer: None,
// message_preparer: None,
// received_keys: Default::default(),
topology: None,
gateway_client: None,
// gateway_client: None,
gateway_identity: None,
on_message: None,
on_gateway_connect: None,
disabled_credentials_mode: true,
@@ -93,54 +115,138 @@ impl NymClient {
self.disabled_credentials_mode = disabled_credentials_mode;
}
fn self_recipient(&self) -> Recipient {
fn as_mix_recipient(&self) -> Recipient {
Recipient::new(
*self.identity.public_key(),
*self.encryption_keys.public_key(),
self.gateway_client
.as_ref()
.expect("gateway connection was not established!")
.gateway_identity(),
*self.key_manager.identity_keypair().public_key(),
*self.key_manager.encryption_keypair().public_key(),
self.gateway_identity
.expect("gateway connection was not established!"),
)
}
pub fn self_address(&self) -> String {
self.self_recipient().to_string()
return "foomp".into();
// self.as_mix_recipient().to_string()
}
// Right now it's impossible to have async exported functions to take `&self` rather than self
pub async fn initial_setup(self) -> Self {
let disabled_credentials_mode = self.disabled_credentials_mode;
// future constantly pumping loop cover traffic at some specified average rate
// the pumped traffic goes to the MixTrafficController
fn start_cover_traffic_stream(
&self,
topology_accessor: TopologyAccessor,
mix_tx: BatchMixMessageSender,
) {
console_log!("Starting loop cover traffic stream...");
let bandwidth_controller = None;
LoopCoverTrafficStream::new(
self.key_manager.ack_key(),
AVERAGE_ACK_DELAY,
AVERAGE_PACKET_DELAY,
LOOP_COVER_STREAM_AVERAGE_DELAY,
mix_tx,
self.as_mix_recipient(),
topology_accessor,
)
.start();
}
let mut client = self.get_and_update_topology().await;
let gateway = client.choose_gateway();
let (mixnet_messages_sender, mixnet_messages_receiver) = mpsc::unbounded();
let (ack_sender, ack_receiver) = mpsc::unbounded();
let mut gateway_client = GatewayClient::new(
gateway.clients_address(),
Arc::clone(&client.identity),
gateway.identity_key,
gateway.owner.clone(),
None,
mixnet_messages_sender,
ack_sender,
DEFAULT_GATEWAY_RESPONSE_TIMEOUT,
bandwidth_controller,
fn start_real_traffic_controller(
&self,
topology_accessor: TopologyAccessor,
ack_receiver: AcknowledgementReceiver,
input_receiver: InputMessageReceiver,
mix_sender: BatchMixMessageSender,
) {
let controller_config = real_messages_control::Config::new(
self.key_manager.ack_key(),
ACK_WAIT_MULTIPLIER,
ACK_WAIT_ADDITION,
AVERAGE_ACK_DELAY,
MESSAGE_STREAM_AVERAGE_DELAY,
AVERAGE_PACKET_DELAY,
self.as_mix_recipient(),
);
gateway_client.set_disabled_credentials_mode(disabled_credentials_mode);
console_log!("Starting real traffic stream...");
RealMessagesController::new(
controller_config,
ack_receiver,
input_receiver,
mix_sender,
topology_accessor,
)
.start();
}
// buffer controlling all messages fetched from provider
// required so that other components would be able to use them (say the websocket)
fn start_received_messages_buffer_controller(
&self,
query_receiver: ReceivedBufferRequestReceiver,
mixnet_receiver: MixnetMessageReceiver,
) {
console_log!("Starting received messages buffer controller...");
ReceivedMessagesBufferController::new(
self.key_manager.encryption_keypair(),
query_receiver,
mixnet_receiver,
)
.start()
}
async fn start_gateway_client(
&mut self,
mixnet_message_sender: MixnetMessageSender,
ack_sender: AcknowledgementSender,
) -> GatewayClient {
let gateway_owner = "n1kymvkx6vsq7pvn6hfurkpg06h3j4gxj4em7tlg".into();
let gateway_id = "E3mvZTHQCdBvhfr178Swx9g4QG3kkRUun7YnToLMcMbM".to_string();
// TODO: might need a port
let gateway_address = "ws://213.219.38.119:9000".into();
// let gateway_address = "213.219.38.119".into();
// for now there are no configs, etc.
// let gateway_id = self.config.get_base().get_gateway_id();
// if gateway_id.is_empty() {
// panic!("The identity of the gateway is unknown - did you run `nym-client` init?")
// }
// let gateway_owner = self.config.get_base().get_gateway_owner();
// if gateway_owner.is_empty() {
// panic!("The owner of the gateway is unknown - did you run `nym-client` init?")
// }
// let gateway_address = self.config.get_base().get_gateway_listener();
// if gateway_address.is_empty() {
// panic!("The address of the gateway is unknown - did you run `nym-client` init?")
// }
let gateway_identity = identity::PublicKey::from_base58_string(gateway_id)
.expect("provided gateway id is invalid!");
self.force_update_internal_topology().await;
let gateway = self.choose_gateway();
let mut gateway_client = GatewayClient::new(
gateway_address,
self.key_manager.identity_keypair(),
gateway_identity,
gateway_owner,
Some(self.key_manager.gateway_shared_key()),
mixnet_message_sender,
ack_sender,
GATEWAY_RESPONSE_TIMEOUT,
None,
);
gateway_client.set_disabled_credentials_mode(self.disabled_credentials_mode);
gateway_client
.authenticate_and_start()
.await
.expect("could not authenticate and start up the gateway connection");
client.gateway_client = Some(gateway_client);
match client.on_gateway_connect.as_ref() {
match self.on_gateway_connect.as_ref() {
Some(callback) => {
callback
.call0(&JsValue::null())
@@ -149,28 +255,168 @@ impl NymClient {
None => console_log!("Gateway connection established - no callback specified"),
};
let rng = rand::rngs::OsRng;
let message_preparer = MessagePreparer::new(
rng,
client.self_recipient(),
DEFAULT_AVERAGE_PACKET_DELAY,
DEFAULT_AVERAGE_ACK_DELAY,
self.gateway_identity = Some(gateway_client.gateway_identity());
gateway_client
}
// future responsible for periodically polling directory server and updating
// the current global view of topology
async fn start_topology_refresher(&mut self, topology_accessor: TopologyAccessor) {
let topology_refresher_config = TopologyRefresherConfig::new(
vec![self.validator_server.clone()],
TOPOLOGY_REFRESH_RATE,
env!("CARGO_PKG_VERSION").to_string(),
);
let mut topology_refresher =
TopologyRefresher::new(topology_refresher_config, topology_accessor);
// before returning, block entire runtime to refresh the current network view so that any
// components depending on topology would see a non-empty view
console_log!("Obtaining initial network topology");
topology_refresher.refresh().await;
let received_processor = ReceivedMessagesProcessor::new(
Arc::clone(&client.encryption_keys),
Arc::clone(&client.ack_key),
);
// TODO: a slightly more graceful termination here
if !topology_refresher.is_topology_routable().await {
panic!(
"The current network topology seem to be insufficient to route any packets through\
- check if enough nodes and a gateway are online"
);
}
client.message_preparer = Some(message_preparer);
console_log!("Starting topology refresher...");
topology_refresher.start();
}
spawn_local(received_processor.start_processing(
// controller for sending sphinx packets to mixnet (either real traffic or cover traffic)
// TODO: if we want to send control messages to gateway_client, this CAN'T take the ownership
// over it. Perhaps GatewayClient needs to be thread-shareable or have some channel for
// requests?
fn start_mix_traffic_controller(
&mut self,
mix_rx: BatchMixMessageReceiver,
gateway_client: GatewayClient,
) {
console_log!("Starting mix traffic controller...");
MixTrafficController::new(mix_rx, gateway_client).start();
}
pub async fn start(mut self) -> NymClient {
// println!("hello world print");
// console_log!("hello world log");
// self
console_log!("Starting nym client");
// channels for inter-component communication
// TODO: make the channels be internally created by the relevant components
// rather than creating them here, so say for example the buffer controller would create the request channels
// and would allow anyone to clone the sender channel
// sphinx_message_sender is the transmitter for any component generating sphinx packets that are to be sent to the mixnet
// they are used by cover traffic stream and real traffic stream
// sphinx_message_receiver is the receiver used by MixTrafficController that sends the actual traffic
let (sphinx_message_sender, sphinx_message_receiver) = mpsc::unbounded();
// unwrapped_sphinx_sender is the transmitter of mixnet messages received from the gateway
// unwrapped_sphinx_receiver is the receiver for said messages - used by ReceivedMessagesBuffer
let (mixnet_messages_sender, mixnet_messages_receiver) = mpsc::unbounded();
// used for announcing connection or disconnection of a channel for pushing re-assembled messages to
let (received_buffer_request_sender, received_buffer_request_receiver) = mpsc::unbounded();
// channels responsible for controlling real messages
let (input_sender, input_receiver) = mpsc::unbounded::<InputMessage>();
// channels responsible for controlling ack messages
let (ack_sender, ack_receiver) = mpsc::unbounded();
let shared_topology_accessor = TopologyAccessor::new();
// the components are started in very specific order. Unless you know what you are doing,
// do not change that.
self.start_topology_refresher(shared_topology_accessor.clone())
.await;
self.start_received_messages_buffer_controller(
received_buffer_request_receiver,
mixnet_messages_receiver,
ack_receiver,
client.on_message.take().expect("on_message was not set!"),
));
);
client
let gateway_client = self
.start_gateway_client(mixnet_messages_sender, ack_sender)
.await;
self.start_mix_traffic_controller(sphinx_message_receiver, gateway_client);
self.start_real_traffic_controller(
shared_topology_accessor.clone(),
ack_receiver,
input_receiver,
sphinx_message_sender.clone(),
);
self.start_cover_traffic_stream(shared_topology_accessor, sphinx_message_sender);
self
}
// Right now it's impossible to have async exported functions to take `&self` rather than self
pub async fn initial_setup(self) -> Self {
// let disabled_credentials_mode = self.disabled_credentials_mode;
//
// let bandwidth_controller = None;
//
// let mut client = self.get_and_update_topology().await;
// let gateway = client.choose_gateway();
//
// let (mixnet_messages_sender, mixnet_messages_receiver) = mpsc::unbounded();
// let (ack_sender, ack_receiver) = mpsc::unbounded();
//
// let mut gateway_client = GatewayClient::new(
// gateway.clients_address(),
// Arc::clone(&client.identity),
// gateway.identity_key,
// gateway.owner.clone(),
// None,
// mixnet_messages_sender,
// ack_sender,
// GATEWAY_RESPONSE_TIMEOUT,
// bandwidth_controller,
// );
//
// gateway_client.set_disabled_credentials_mode(disabled_credentials_mode);
//
// gateway_client
// .authenticate_and_start()
// .await
// .expect("could not authenticate and start up the gateway connection");
//
// client.gateway_client = Some(gateway_client);
// match client.on_gateway_connect.as_ref() {
// Some(callback) => {
// callback
// .call0(&JsValue::null())
// .expect("on connect callback failed!");
// }
// None => console_log!("Gateway connection established - no callback specified"),
// };
//
// let rng = rand::rngs::OsRng;
// let message_preparer = MessagePreparer::new(
// rng,
// client.self_recipient(),
// AVERAGE_PACKET_DELAY,
// AVERAGE_ACK_DELAY,
// );
//
// let received_processor = ReceivedMessagesProcessor::new(
// Arc::clone(&client.encryption_keys),
// Arc::clone(&client.ack_key),
// );
//
// client.message_preparer = Some(message_preparer);
//
// spawn_local(received_processor.start_processing(
// mixnet_messages_receiver,
// ack_receiver,
// client.on_message.take().expect("on_message was not set!"),
// ));
//
self
}
// Right now it's impossible to have async exported functions to take `&mut self` rather than mut self
@@ -178,37 +424,39 @@ impl NymClient {
pub async fn send_message(mut self, message: String, recipient: String) -> Self {
console_log!("Sending {} to {}", message, recipient);
let message_bytes = message.into_bytes();
let recipient = Recipient::try_from_base58_string(recipient).unwrap();
todo!()
let topology = self
.topology
.as_ref()
.expect("did not obtain topology before");
let message_preparer = self.message_preparer.as_mut().unwrap();
let (split_message, _reply_keys) = message_preparer
.prepare_and_split_message(message_bytes, false, topology)
.expect("failed to split the message");
let mut mix_packets = Vec::with_capacity(split_message.len());
for message_chunk in split_message {
// don't bother with acks etc. for time being
let prepared_fragment = message_preparer
.prepare_chunk_for_sending(message_chunk, topology, &self.ack_key, &recipient)
.unwrap();
console_warn!("packet is going to have round trip time of {:?}, but we're not going to do anything for acks anyway ", prepared_fragment.total_delay);
mix_packets.push(prepared_fragment.mix_packet);
}
self.gateway_client
.as_mut()
.unwrap()
.batch_send_mix_packets(mix_packets)
.await
.unwrap();
self
// let message_bytes = message.into_bytes();
// let recipient = Recipient::try_from_base58_string(recipient).unwrap();
//
// let topology = self
// .topology
// .as_ref()
// .expect("did not obtain topology before");
//
// let message_preparer = self.message_preparer.as_mut().unwrap();
//
// let (split_message, _reply_keys) = message_preparer
// .prepare_and_split_message(message_bytes, false, topology)
// .expect("failed to split the message");
//
// let mut mix_packets = Vec::with_capacity(split_message.len());
// for message_chunk in split_message {
// // don't bother with acks etc. for time being
// let prepared_fragment = message_preparer
// .prepare_chunk_for_sending(message_chunk, topology, &self.ack_key, &recipient)
// .unwrap();
//
// console_warn!("packet is going to have round trip time of {:?}, but we're not going to do anything for acks anyway ", prepared_fragment.total_delay);
// mix_packets.push(prepared_fragment.mix_packet);
// }
// self.gateway_client
// .as_mut()
// .unwrap()
// .batch_send_mix_packets(mix_packets)
// .await
// .unwrap();
// self
}
pub(crate) fn choose_gateway(&self) -> &gateway::Node {
@@ -226,31 +474,35 @@ impl NymClient {
// self: Rc<Self>
// or this: Rc<RefCell<Self>>
pub async fn get_and_update_topology(mut self) -> Self {
self.force_update_internal_topology().await;
self
}
pub(crate) async fn force_update_internal_topology(&mut self) {
let new_topology = self.get_nym_topology().await;
self.update_topology(new_topology);
self
}
pub(crate) fn update_topology(&mut self, topology: NymTopology) {
self.topology = Some(topology)
}
// when updated to 0.10.0, to prevent headache later on, this function requires those two imports:
// use js_sys::Promise;
// use wasm_bindgen_futures::future_to_promise;
//
// pub fn get_full_topology_json(&self) -> Promise {
// let validator_client_config = validator_client::Config::new(
// vec![self.validator_server.clone()],
// &self.mixnet_contract_address,
// );
// let validator_client = validator_client::Client::new(validator_client_config);
//
// future_to_promise(async move {
// let topology = &validator_client.get_active_topology().await.unwrap();
// Ok(JsValue::from_serde(&topology).unwrap())
// })
// }
// // when updated to 0.10.0, to prevent headache later on, this function requires those two imports:
// // use js_sys::Promise;
// // use wasm_bindgen_futures::future_to_promise;
// //
// // pub fn get_full_topology_json(&self) -> Promise {
// // let validator_client_config = validator_client::Config::new(
// // vec![self.validator_server.clone()],
// // &self.mixnet_contract_address,
// // );
// // let validator_client = validator_client::Client::new(validator_client_config);
// //
// // future_to_promise(async move {
// // let topology = &validator_client.get_active_topology().await.unwrap();
// // Ok(JsValue::from_serde(&topology).unwrap())
// // })
// // }
pub(crate) async fn get_nym_topology(&self) -> NymTopology {
let validator_client = validator_client::ApiClient::new(self.validator_server.clone());
common/client-libs/gateway-client/Cargo.toml
+1 -1
View File
@@ -34,7 +34,7 @@ default-features = false
# non-wasm-only dependencies
[target."cfg(not(target_arch = \"wasm32\"))".dependencies.tokio]
version = "1.19.1"
version = "1.21.2"
features = ["macros", "rt", "net", "sync", "time"]
[target."cfg(not(target_arch = \"wasm32\"))".dependencies.tokio-stream]
common/client-libs/mixnet-client/Cargo.toml
+1 -1
View File
@@ -9,7 +9,7 @@ edition = "2021"
[dependencies]
futures = "0.3"
log = "0.4.8"
tokio = { version = "1.19.1", features = ["time", "net", "rt"] }
tokio = { version = "1.21.2", features = ["time", "net", "rt"] }
tokio-util = { version = "0.7.3", features = ["codec"] }
# internal
common/client-libs/validator-client/Cargo.toml
+1 -1
View File
@@ -22,7 +22,7 @@ reqwest = { version = "0.11", features = ["json"] }
thiserror = "1"
log = "0.4"
url = { version = "2.2", features = ["serde"] }
tokio = { version = "1.19.1", features = ["sync", "time"] }
tokio = { version = "1.21.2", features = ["sync", "time"] }
futures = "0.3"
coconut-interface = { path = "../../coconut-interface" }
common/credential-storage/Cargo.toml
+2 -2
View File
@@ -11,9 +11,9 @@ async-trait = { version = "0.1.51" }
log = "0.4"
sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"]}
thiserror = "1.0"
tokio = { version = "1.19.1", features = [ "rt-multi-thread", "net", "signal", "fs" ] }
tokio = { version = "1.21.2", features = [ "rt-multi-thread", "net", "signal", "fs" ] }
[build-dependencies]
sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] }
tokio = { version = "1.19.1", features = ["rt-multi-thread", "macros"] }
tokio = { version = "1.21.2", features = ["rt-multi-thread", "macros"] }
common/mixnode-common/Cargo.toml
+1 -1
View File
@@ -13,7 +13,7 @@ humantime-serde = "1.0"
log = "0.4"
rand = "0.8"
serde = { version = "1.0", features = ["derive"] }
tokio = { version = "1.19.1", features = ["time", "macros", "rt", "net", "io-util"] }
tokio = { version = "1.21.2", features = ["time", "macros", "rt", "net", "io-util"] }
tokio-util = { version = "0.7.3", features = ["codec"] }
url = "2.2"
common/nonexhaustive-delayqueue/Cargo.toml
+1 -1
View File
@@ -7,6 +7,6 @@ edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
tokio = { version = "1.19.1", features = [] }
tokio = { version = "1.21.2", features = [] }
tokio-stream = "0.1.9" # this one seems to be a thing until `Stream` trait is stabilised in stdlib
tokio-util = { version = "0.7.3", features = ["time"] }
common/nymsphinx/Cargo.toml
+1 -1
View File
@@ -33,5 +33,5 @@ mixnet-contract-common = { path = "../cosmwasm-smart-contracts/mixnet-contract"
path = "framing"
[target."cfg(not(target_arch = \"wasm32\"))".dependencies.tokio]
version = "1.19.1"
version = "1.21.2"
features = ["sync"]
common/nymsphinx/src/preparer/mod.rs
+2 -1
View File
@@ -53,7 +53,8 @@ impl From<NymTopologyError> for PreparationError {
/// Prepares the message that is to be sent through the mix network by attaching
/// an optional reply-SURB, padding it to appropriate length, encrypting its content,
/// and chunking into appropriate size [`Fragment`]s.
#[cfg_attr(not(target_arch = "wasm32"), derive(Clone))]
// #[cfg_attr(not(target_arch = "wasm32"), derive(Clone))]
#[derive(Clone)]
#[must_use]
pub struct MessagePreparer<R: CryptoRng + Rng> {
/// Instance of a cryptographically secure random number generator.
common/socks5/proxy-helpers/Cargo.toml
+1 -1
View File
@@ -8,7 +8,7 @@ edition = "2021"
[dependencies]
bytes = "1.0"
tokio = { version = "1.19.1", features = [ "net", "io-util", "sync", "macros", "time", "rt-multi-thread" ] }
tokio = { version = "1.21.2", features = [ "net", "io-util", "sync", "macros", "time", "rt-multi-thread" ] }
tokio-util = { version = "0.7.3", features = [ "io" ] } # reason for getting this guy is to to able to port to tokio 1.X more quickly by being able to use
# their `read_buf` [from the util crate] replacement rather than having to rethink/reimplement `AvailableReader` with the new AsyncRead trait definition.
# In the long run, the dependency should probably get removed in favour of pure-tokio implementation, but for time being it's fine.
common/statistics/Cargo.toml
+1 -1
View File
@@ -16,4 +16,4 @@ serde = { version = "1.0", features = ["derive"] }
serde_json = "1"
sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "chrono"]}
thiserror = "1"
tokio = { version = "1.19.1", features = [ "time" ] }
tokio = { version = "1.21.2", features = [ "time" ] }
common/task/Cargo.toml
+2 -2
View File
@@ -7,7 +7,7 @@ edition = "2021"
[dependencies]
log = "0.4"
tokio = { version = "1.19.1", features = ["rt-multi-thread", "net", "signal"] }
tokio = { version = "1.21.2", features = ["macros", "signal", "time", "sync"] }
[dev-dependencies]
tokio = { version = "1.19.1", features = ["rt-multi-thread", "net", "signal", "test-util", "macros"] }
tokio = { version = "1.21.2", features = ["rt-multi-thread", "net", "signal", "test-util", "macros"] }
explorer-api/Cargo.toml
+5
View File
@@ -22,7 +22,12 @@ schemars = { version = "0.8", features = ["preserve_order"] }
serde = "1.0.126"
serde_json = "1.0.66"
thiserror = "1.0.29"
<<<<<<< Updated upstream
tokio = {version = "1.19.1", features = ["full"] }
=======
tokio = {version = "1.21.2", features = ["full"] }
maxminddb = "0.23.0"
>>>>>>> Stashed changes
mixnet-contract-common = { path = "../common/cosmwasm-smart-contracts/mixnet-contract" }
network-defaults = { path = "../common/network-defaults" }
gateway/Cargo.toml
+2 -2
View File
@@ -39,7 +39,7 @@ sqlx = { version = "0.5", features = [
] }
subtle-encoding = { version = "0.5", features = ["bech32-preview"] }
thiserror = "1"
tokio = { version = "1.19.1", features = [
tokio = { version = "1.21.2", features = [
"rt-multi-thread",
"net",
"signal",
@@ -77,7 +77,7 @@ coconut = [
]
[build-dependencies]
tokio = { version = "1.19.1", features = ["rt-multi-thread", "macros"] }
tokio = { version = "1.21.2", features = ["rt-multi-thread", "macros"] }
sqlx = { version = "0.5", features = [
"runtime-tokio-rustls",
"sqlite",
mixnode/Cargo.toml
+2 -2
View File
@@ -32,7 +32,7 @@ rand = "0.7.3"
rocket = { version = "0.5.0-rc.2", features = ["json"] }
serde = { version="1.0", features = ["derive"] }
sysinfo = "0.24.1"
tokio = { version="1.19.1", features = ["rt-multi-thread", "net", "signal"] }
tokio = { version="1.21.2", features = ["rt-multi-thread", "net", "signal"] }
tokio-util = { version="0.7.3", features = ["codec"] }
toml = "0.5.8"
url = { version = "2.2", features = ["serde"] }
@@ -51,7 +51,7 @@ validator-client = { path="../common/client-libs/validator-client" }
version-checker = { path="../common/version-checker" }
[dev-dependencies]
tokio = { version="1.19.1", features = ["rt-multi-thread", "net", "signal", "test-util"] }
tokio = { version="1.21.2", features = ["rt-multi-thread", "net", "signal", "test-util"] }
nymsphinx-types = { path = "../common/nymsphinx/types" }
nymsphinx-params = { path = "../common/nymsphinx/params" }
nym-connect/src-tauri/Cargo.toml
+1 -1
View File
@@ -35,7 +35,7 @@ tap = "1.0.1"
tauri = { version = "^1.0.2", features = ["clipboard-write-text", "shell-open", "system-tray", "updater"] }
tendermint-rpc = "0.23.0"
thiserror = "1.0"
tokio = { version = "1.19.1", features = ["sync", "time"] }
tokio = { version = "1.21.2", features = ["sync", "time"] }
url = "2.2"
client-core = { path = "../../clients/client-core" }
nym-wallet/Cargo.lock
Generated
+8 -28
View File
@@ -2389,9 +2389,9 @@ checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646"
[[package]]
name = "libc"
version = "0.2.119"
version = "0.2.134"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1bf2e165bb3457c8e098ea76f3e3bc9db55f87aa90d52d0e6be741470916aaa4"
checksum = "329c933548736bc49fd575ee68c89e8be4d260064184389a5b77517cddd99ffb"
[[package]]
name = "line-wrap"
@@ -2540,25 +2540,14 @@ dependencies = [
[[package]]
name = "mio"
version = "0.8.2"
version = "0.8.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "52da4364ffb0e4fe33a9841a98a3f3014fb964045ce4f7a45a398243c8d6b0c9"
checksum = "57ee1c23c7c63b0c9250c339ffdc69255f110b298b901b9f6c82547b7b87caaf"
dependencies = [
"libc",
"log",
"miow",
"ntapi",
"wasi 0.11.0+wasi-snapshot-preview1",
"winapi",
]
[[package]]
name = "miow"
version = "0.3.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b9f1c5b025cda876f66ef43a113f91ebc9f4ccef34843000e0adf6ebbab84e21"
dependencies = [
"winapi",
"windows-sys",
]
[[package]]
@@ -2660,15 +2649,6 @@ version = "0.1.14"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "72ef4a56884ca558e5ddb05a1d1e7e1bfd9a68d9ed024c21704cc98872dae1bb"
[[package]]
name = "ntapi"
version = "0.3.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c28774a7fd2fbb4f0babd8237ce554b73af68021b5f695a3cebd6c59bac0980f"
dependencies = [
"winapi",
]
[[package]]
name = "num-derive"
version = "0.3.3"
@@ -4969,16 +4949,16 @@ checksum = "25eb0ca3468fc0acc11828786797f6ef9aa1555e4a211a60d64cc8e4d1be47d6"
[[package]]
name = "tokio"
version = "1.19.2"
version = "1.21.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c51a52ed6686dd62c320f9b89299e9dfb46f730c7a48e635c19f21d116cb1439"
checksum = "a9e03c497dc955702ba729190dc4aac6f2a0ce97f913e5b1b5912fc5039d9099"
dependencies = [
"autocfg 1.1.0",
"bytes",
"libc",
"memchr",
"mio",
"num_cpus",
"once_cell",
"parking_lot 0.12.1",
"pin-project-lite",
"signal-hook-registry",
service-providers/network-requester/Cargo.toml
+1 -1
View File
@@ -23,7 +23,7 @@ reqwest = { version = "0.11.11", features = ["json"] }
serde = { version = "1.0", features = ["derive"] }
sqlx = { version = "0.6.1", features = ["runtime-tokio-rustls", "chrono"]}
thiserror = "1.0"
tokio = { version = "1.19", features = [ "net", "rt-multi-thread", "macros" ] }
tokio = { version = "1.21.2", features = [ "net", "rt-multi-thread", "macros" ] }
tokio-tungstenite = "0.17.2"
validator-api/Cargo.toml
+2 -2
View File
@@ -36,7 +36,7 @@ serde_json = "1.0"
tap = "1.0"
thiserror = "1.0"
time = { version = "0.3.14", features = ["serde-human-readable", "parsing"] }
tokio = { version = "1.19", features = [
tokio = { version = "1.21.2", features = [
"rt-multi-thread",
"macros",
"signal",
@@ -96,7 +96,7 @@ no-reward = []
generate-ts = ["ts-rs"]
[build-dependencies]
tokio = { version = "1.19.1", features = ["rt-multi-thread", "macros"] }
tokio = { version = "1.21.2", features = ["rt-multi-thread", "macros"] }
sqlx = { version = "0.6.2", features = [
"runtime-tokio-rustls",
"sqlite",