Merge pull request #6099 from nymtech/bugfix/incompatibility-fixes
Bugfix/incompatibility fixes
This commit is contained in:
@@ -60,6 +60,7 @@ impl SocketClient {
|
||||
let ClientInput {
|
||||
connection_command_sender,
|
||||
input_sender,
|
||||
..
|
||||
} = client_input;
|
||||
|
||||
let ClientOutput {
|
||||
|
||||
@@ -87,6 +87,7 @@ pub mod storage;
|
||||
pub struct ClientInput {
|
||||
pub connection_command_sender: ConnectionCommandSender,
|
||||
pub input_sender: InputMessageSender,
|
||||
pub client_request_sender: ClientRequestSender,
|
||||
}
|
||||
|
||||
impl ClientInput {
|
||||
@@ -746,9 +747,12 @@ where
|
||||
shutdown_tracker: &ShutdownTracker,
|
||||
) -> (BatchMixMessageSender, ClientRequestSender) {
|
||||
info!("Starting mix traffic controller...");
|
||||
let (mut mix_traffic_controller, mix_tx, client_tx) =
|
||||
let mut mix_traffic_controller =
|
||||
MixTrafficController::new(gateway_transceiver, shutdown_tracker.clone_shutdown_token());
|
||||
|
||||
let mix_tx = mix_traffic_controller.mix_rx();
|
||||
let client_tx = mix_traffic_controller.client_tx();
|
||||
|
||||
shutdown_tracker.try_spawn_named(
|
||||
async move { mix_traffic_controller.run().await },
|
||||
"MixTrafficController",
|
||||
@@ -1074,6 +1078,7 @@ where
|
||||
client_input: ClientInput {
|
||||
connection_command_sender: client_connection_tx,
|
||||
input_sender,
|
||||
client_request_sender,
|
||||
},
|
||||
},
|
||||
client_output: ClientOutputStatus::AwaitingConsumer {
|
||||
@@ -1089,7 +1094,6 @@ where
|
||||
},
|
||||
stats_reporter,
|
||||
shutdown_handle: shutdown_tracker, // The primary tracker for this client
|
||||
client_request_sender,
|
||||
forget_me: self.config.debug.forget_me,
|
||||
remember_me: self.config.debug.remember_me,
|
||||
})
|
||||
@@ -1103,7 +1107,6 @@ pub struct BaseClient {
|
||||
pub client_output: ClientOutputStatus,
|
||||
pub client_state: ClientState,
|
||||
pub stats_reporter: ClientStatsSender,
|
||||
pub client_request_sender: ClientRequestSender,
|
||||
pub shutdown_handle: ShutdownTracker,
|
||||
pub forget_me: ForgetMe,
|
||||
pub remember_me: RememberMe,
|
||||
|
||||
@@ -25,8 +25,10 @@ pub struct Empty;
|
||||
pub struct MixTrafficController {
|
||||
gateway_transceiver: Box<dyn GatewayTransceiver + Send>,
|
||||
|
||||
mix_tx: BatchMixMessageSender,
|
||||
mix_rx: BatchMixMessageReceiver,
|
||||
client_rx: ClientRequestReceiver,
|
||||
client_tx: ClientRequestSender,
|
||||
|
||||
// TODO: this is temporary work-around.
|
||||
// in long run `gateway_client` will be moved away from `MixTrafficController` anyway.
|
||||
@@ -36,14 +38,7 @@ pub struct MixTrafficController {
|
||||
}
|
||||
|
||||
impl MixTrafficController {
|
||||
pub fn new<T>(
|
||||
gateway_transceiver: T,
|
||||
shutdown_token: ShutdownToken,
|
||||
) -> (
|
||||
MixTrafficController,
|
||||
BatchMixMessageSender,
|
||||
ClientRequestSender,
|
||||
)
|
||||
pub fn new<T>(gateway_transceiver: T, shutdown_token: ShutdownToken) -> MixTrafficController
|
||||
where
|
||||
T: GatewayTransceiver + Send + 'static,
|
||||
{
|
||||
@@ -52,41 +47,30 @@ impl MixTrafficController {
|
||||
|
||||
let (client_sender, client_receiver) = tokio::sync::mpsc::channel(8);
|
||||
|
||||
(
|
||||
MixTrafficController {
|
||||
gateway_transceiver: Box::new(gateway_transceiver),
|
||||
mix_rx: message_receiver,
|
||||
client_rx: client_receiver,
|
||||
consecutive_gateway_failure_count: 0,
|
||||
shutdown_token,
|
||||
},
|
||||
message_sender,
|
||||
client_sender,
|
||||
)
|
||||
MixTrafficController {
|
||||
gateway_transceiver: Box::new(gateway_transceiver),
|
||||
mix_tx: message_sender,
|
||||
mix_rx: message_receiver,
|
||||
client_rx: client_receiver,
|
||||
client_tx: client_sender,
|
||||
consecutive_gateway_failure_count: 0,
|
||||
shutdown_token,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn new_dynamic(
|
||||
gateway_transceiver: Box<dyn GatewayTransceiver + Send>,
|
||||
shutdown_token: ShutdownToken,
|
||||
) -> (
|
||||
MixTrafficController,
|
||||
BatchMixMessageSender,
|
||||
ClientRequestSender,
|
||||
) {
|
||||
let (message_sender, message_receiver) =
|
||||
tokio::sync::mpsc::channel(MIX_MESSAGE_RECEIVER_BUFFER_SIZE);
|
||||
let (client_sender, client_receiver) = tokio::sync::mpsc::channel(8);
|
||||
(
|
||||
MixTrafficController {
|
||||
gateway_transceiver,
|
||||
mix_rx: message_receiver,
|
||||
client_rx: client_receiver,
|
||||
consecutive_gateway_failure_count: 0,
|
||||
shutdown_token,
|
||||
},
|
||||
message_sender,
|
||||
client_sender,
|
||||
)
|
||||
) -> MixTrafficController {
|
||||
Self::new(gateway_transceiver, shutdown_token)
|
||||
}
|
||||
|
||||
pub fn client_tx(&self) -> ClientRequestSender {
|
||||
self.client_tx.clone()
|
||||
}
|
||||
|
||||
pub fn mix_rx(&self) -> BatchMixMessageSender {
|
||||
self.mix_tx.clone()
|
||||
}
|
||||
|
||||
async fn on_messages(
|
||||
@@ -145,34 +129,26 @@ impl MixTrafficController {
|
||||
trace!("MixTrafficController: Received shutdown");
|
||||
break;
|
||||
}
|
||||
mix_packets = self.mix_rx.recv() => match mix_packets {
|
||||
Some(mix_packets) => {
|
||||
if let Err(err) = self.on_messages(mix_packets).await {
|
||||
error!("Failed to send sphinx packet(s) to the gateway: {err}");
|
||||
if self.consecutive_gateway_failure_count == MAX_FAILURE_COUNT {
|
||||
// Disconnect from the gateway. If we should try to re-connect
|
||||
// is handled at a higher layer.
|
||||
error!("Failed to send sphinx packet to the gateway {MAX_FAILURE_COUNT} times in a row - assuming the gateway is dead");
|
||||
// Do we need to handle the embedded mixnet client case
|
||||
// separately?
|
||||
break;
|
||||
}
|
||||
// mix_rx should never error out as we're holding one instance of the sender
|
||||
|
||||
Some(mix_packets) = self.mix_rx.recv() => {
|
||||
if let Err(err) = self.on_messages(mix_packets).await {
|
||||
error!("Failed to send sphinx packet(s) to the gateway: {err}");
|
||||
if self.consecutive_gateway_failure_count == MAX_FAILURE_COUNT {
|
||||
// Disconnect from the gateway. If we should try to re-connect
|
||||
// is handled at a higher layer.
|
||||
error!("Failed to send sphinx packet to the gateway {MAX_FAILURE_COUNT} times in a row - assuming the gateway is dead");
|
||||
// Do we need to handle the embedded mixnet client case
|
||||
// separately?
|
||||
self.shutdown_token.cancel();
|
||||
break;
|
||||
}
|
||||
},
|
||||
None => {
|
||||
trace!("MixTrafficController: Stopping since channel closed");
|
||||
break;
|
||||
}
|
||||
},
|
||||
client_request = self.client_rx.recv() => match client_request {
|
||||
Some(client_request) => {
|
||||
self.on_client_request(client_request).await;
|
||||
},
|
||||
None => {
|
||||
trace!("MixTrafficController, client request channel closed");
|
||||
break
|
||||
}
|
||||
},
|
||||
// client_rx should never error out as we're holding one instance of the sender
|
||||
Some(client_request) = self.client_rx.recv() => {
|
||||
self.on_client_request(client_request).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
debug!("MixTrafficController: Exiting");
|
||||
|
||||
@@ -119,6 +119,7 @@ where
|
||||
let ClientInput {
|
||||
connection_command_sender,
|
||||
input_sender,
|
||||
..
|
||||
} = client_input;
|
||||
|
||||
let ClientOutput {
|
||||
|
||||
@@ -124,6 +124,7 @@ pub struct Config {
|
||||
pub node_status_api: NodeStatusAPI,
|
||||
|
||||
#[serde(alias = "topology_cacher")]
|
||||
#[serde(default)]
|
||||
pub describe_cache: DescribeCache,
|
||||
|
||||
#[serde(default)]
|
||||
|
||||
@@ -70,30 +70,6 @@ per_node_test_packets = {{ network_monitor.debug.per_node_test_packets }}
|
||||
# Path to the database file containing uptime statuses for all mixnodes and gateways.
|
||||
database_path = '{{ node_status_api.storage_paths.database_path }}'
|
||||
|
||||
[node_status_api.debug]
|
||||
|
||||
caching_interval = '{{ node_status_api.debug.caching_interval }}'
|
||||
|
||||
|
||||
##### topology cacher config options #####
|
||||
|
||||
[topology_cacher.debug]
|
||||
|
||||
caching_interval = '{{ topology_cacher.debug.caching_interval }}'
|
||||
|
||||
|
||||
##### circulating supply cacher config options #####
|
||||
|
||||
[circulating_supply_cacher]
|
||||
|
||||
# Specifies whether circulating supply caching service is enabled in this process.
|
||||
enabled = {{ circulating_supply_cacher.enabled }}
|
||||
|
||||
[circulating_supply_cacher.debug]
|
||||
|
||||
caching_interval = '{{ circulating_supply_cacher.debug.caching_interval }}'
|
||||
|
||||
|
||||
##### rewarding config options #####
|
||||
|
||||
[rewarding]
|
||||
|
||||
@@ -850,7 +850,6 @@ where
|
||||
stats_events_reporter,
|
||||
started_client.shutdown_handle,
|
||||
None,
|
||||
started_client.client_request_sender,
|
||||
started_client.forget_me,
|
||||
started_client.remember_me,
|
||||
))
|
||||
|
||||
@@ -58,7 +58,6 @@ pub struct MixnetClient {
|
||||
|
||||
// internal state used for the `Stream` implementation
|
||||
_buffered: Vec<ReconstructedMessage>,
|
||||
pub(crate) client_request_sender: ClientRequestSender,
|
||||
pub(crate) forget_me: ForgetMe,
|
||||
pub(crate) remember_me: RememberMe,
|
||||
}
|
||||
@@ -75,7 +74,6 @@ impl MixnetClient {
|
||||
stats_events_reporter: ClientStatsSender,
|
||||
task_handle: ShutdownTracker,
|
||||
packet_type: Option<PacketType>,
|
||||
client_request_sender: ClientRequestSender,
|
||||
forget_me: ForgetMe,
|
||||
remember_me: RememberMe,
|
||||
) -> Self {
|
||||
@@ -90,7 +88,6 @@ impl MixnetClient {
|
||||
shutdown_handle: task_handle,
|
||||
packet_type,
|
||||
_buffered: Vec::new(),
|
||||
client_request_sender,
|
||||
forget_me,
|
||||
remember_me,
|
||||
}
|
||||
@@ -129,7 +126,7 @@ impl MixnetClient {
|
||||
}
|
||||
|
||||
pub fn client_request_sender(&self) -> ClientRequestSender {
|
||||
self.client_request_sender.clone()
|
||||
self.client_input.client_request_sender.clone()
|
||||
}
|
||||
|
||||
/// Get the client's identity keys.
|
||||
@@ -252,7 +249,12 @@ impl MixnetClient {
|
||||
client: self.forget_me.client(),
|
||||
stats: self.forget_me.stats(),
|
||||
};
|
||||
match self.client_request_sender.send(client_request).await {
|
||||
match self
|
||||
.client_input
|
||||
.client_request_sender
|
||||
.send(client_request)
|
||||
.await
|
||||
{
|
||||
Ok(_) => Ok(()),
|
||||
Err(e) => {
|
||||
error!("Failed to send forget me request: {e}");
|
||||
@@ -265,7 +267,12 @@ impl MixnetClient {
|
||||
let client_request = ClientRequest::RememberMe {
|
||||
session_type: self.remember_me.session_type(),
|
||||
};
|
||||
match self.client_request_sender.send(client_request).await {
|
||||
match self
|
||||
.client_input
|
||||
.client_request_sender
|
||||
.send(client_request)
|
||||
.await
|
||||
{
|
||||
Ok(_) => Ok(()),
|
||||
Err(e) => {
|
||||
error!("Failed to send forget me request: {e}");
|
||||
|
||||
@@ -1,2 +1,18 @@
|
||||
build-bypass-contract:
|
||||
$(MAKE) -C dkg-bypass-contract build
|
||||
$(MAKE) -C dkg-bypass-contract build
|
||||
|
||||
|
||||
COSMWASM_OPTIMIZER_IMAGE ?= cosmwasm/optimizer:0.17.0
|
||||
COSMWASM_OPTIMIZER_PLATFORM ?= linux/amd64
|
||||
|
||||
build-bypass-contract-docker:
|
||||
docker volume rm nym_contracts_cache 2>/dev/null || true
|
||||
docker volume rm registry_cache 2>/dev/null || true
|
||||
docker run --rm --platform $(COSMWASM_OPTIMIZER_PLATFORM) \
|
||||
-v $(CURDIR)/../../..:/code \
|
||||
--mount type=volume,source=nym_contracts_cache,target=/target \
|
||||
--mount type=volume,source=registry_cache,target=/usr/local/cargo/registry \
|
||||
-e CARGO_BUILD_INCREMENTAL=false \
|
||||
-e RUSTFLAGS="-C target-cpu=generic -C debuginfo=0" \
|
||||
-e SOURCE_DATE_EPOCH=1 \
|
||||
$(COSMWASM_OPTIMIZER_IMAGE) "tools/internal/testnet-manager/dkg-bypass-contract"; \
|
||||
|
||||
@@ -5,6 +5,7 @@ use crate::msg::MigrateMsg;
|
||||
use cosmwasm_schema::cw_serde;
|
||||
use cosmwasm_std::{
|
||||
Addr, Deps, DepsMut, Env, MessageInfo, QueryResponse, Response, StdError, StdResult, Storage,
|
||||
entry_point,
|
||||
};
|
||||
use cw_storage_plus::{Index, IndexList, IndexedMap, Item, Map, MultiIndex};
|
||||
use nym_coconut_dkg_common::dealer::DealerRegistrationDetails;
|
||||
@@ -57,7 +58,9 @@ pub(crate) fn next_node_index(store: &mut dyn Storage) -> StdResult<NodeIndex> {
|
||||
#[cw_serde]
|
||||
pub enum EmptyMessage {}
|
||||
|
||||
#[cfg_attr(not(feature = "library"), cosmwasm_std::entry_point)]
|
||||
// #[cfg_attr(not(feature = "library"), cosmwasm_std::entry_point)]
|
||||
|
||||
#[entry_point]
|
||||
pub fn instantiate(
|
||||
_: DepsMut<'_>,
|
||||
_: Env,
|
||||
@@ -68,7 +71,8 @@ pub fn instantiate(
|
||||
}
|
||||
|
||||
/// Handle an incoming message
|
||||
#[cfg_attr(not(feature = "library"), cosmwasm_std::entry_point)]
|
||||
// #[cfg_attr(not(feature = "library"), cosmwasm_std::entry_point)]
|
||||
#[entry_point]
|
||||
pub fn execute(
|
||||
_: DepsMut<'_>,
|
||||
_: Env,
|
||||
@@ -78,13 +82,15 @@ pub fn execute(
|
||||
Ok(Response::new())
|
||||
}
|
||||
|
||||
#[cfg_attr(not(feature = "library"), cosmwasm_std::entry_point)]
|
||||
// #[cfg_attr(not(feature = "library"), cosmwasm_std::entry_point)]
|
||||
#[entry_point]
|
||||
pub fn query(_: Deps<'_>, _: Env, _: EmptyMessage) -> Result<QueryResponse, StdError> {
|
||||
Ok(Default::default())
|
||||
}
|
||||
|
||||
// LIMITATION: we're not storing dealings themselves
|
||||
#[cfg_attr(not(feature = "library"), cosmwasm_std::entry_point)]
|
||||
// #[cfg_attr(not(feature = "library"), cosmwasm_std::entry_point)]
|
||||
#[entry_point]
|
||||
pub fn migrate(deps: DepsMut<'_>, env: Env, msg: MigrateMsg) -> Result<Response, StdError> {
|
||||
// on migration immediately attempt to rewrite the storage
|
||||
let threshold = (2 * msg.dealers.len() as u64).div_ceil(3);
|
||||
|
||||
@@ -21,6 +21,7 @@ use std::path::Path;
|
||||
use std::time::Duration;
|
||||
use time::OffsetDateTime;
|
||||
use time::format_description::well_known::Rfc3339;
|
||||
use tracing::error;
|
||||
use url::Url;
|
||||
|
||||
struct InitCtx {
|
||||
@@ -682,9 +683,14 @@ impl NetworkManager {
|
||||
})?;
|
||||
|
||||
let now = OffsetDateTime::now_utc();
|
||||
// SAFETY: all the information saved in our contracts should be well-formed
|
||||
let commit_timestamp = OffsetDateTime::parse(&build_info.commit_timestamp, &Rfc3339)
|
||||
.expect("malformed commit timestamp");
|
||||
.inspect_err(|err| {
|
||||
error!(
|
||||
"failed to parse contract build information: {err}. set timestamp was: {}",
|
||||
build_info.commit_timestamp
|
||||
)
|
||||
})
|
||||
.unwrap_or(OffsetDateTime::UNIX_EPOCH);
|
||||
|
||||
let age = now - commit_timestamp;
|
||||
|
||||
|
||||
@@ -66,11 +66,6 @@ pub struct NymClient {
|
||||
_task_manager: ShutdownTracker,
|
||||
|
||||
packet_type: PacketType,
|
||||
|
||||
// We need this to keep the client_request channel alive and avoid jamming up the
|
||||
// JS runtime when the MixTrafficController then tries to reconnect it if it dies
|
||||
#[allow(dead_code)]
|
||||
pub(crate) client_request_sender: ClientRequestSender,
|
||||
}
|
||||
|
||||
// TODO: we don't really need a builder anymore,
|
||||
@@ -263,7 +258,6 @@ impl NymClientBuilder {
|
||||
_full_topology: None,
|
||||
_task_manager: started_client.shutdown_handle,
|
||||
packet_type,
|
||||
client_request_sender: started_client.client_request_sender,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user