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

Merge branch 'develop' into 348-bonding-settings

Browse Source
This commit is contained in:
Gala
2022-09-02 09:39:21 +02:00
parent 1cdca7bec3 a2c6abd3dd
commit 25df7bcd4d
30 changed files with 388 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+4 -1
View File
@@ -53,7 +53,7 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https://
- validator: fixed local docker-compose setup to work on Apple M1 ([#1329])
- explorer-api: listen out for SIGTERM and SIGQUIT too, making it play nicely as a system service ([#1482]).
- network-requester: fix filter for suffix-only domains ([#1487])
- validator-api: listen out for SIGTERM and SIGQUIT too, making it play nicely as a system service ([#1496]).
- validator-api: listen out for SIGTERM and SIGQUIT too, making it play nicely as a system service; cleaner shutdown, without panics ([#1496], [#1573]).
### Changed
@@ -70,6 +70,7 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https://
- gateway, network-statistics: include gateway id in the sent statistical data ([#1478])
- network explorer: tweak how active set probability is shown ([#1503])
- validator-api: rewarder set update fails without panicking on possible nymd queries ([#1520])
- network-requester, socks5 client (nym-connect): send and receive respectively a message error to be displayed about filter check failure ([#1576])
[#1249]: https://github.com/nymtech/nym/pull/1249
@@ -99,6 +100,8 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https://
[#1496]: https://github.com/nymtech/nym/pull/1496
[#1503]: https://github.com/nymtech/nym/pull/1503
[#1520]: https://github.com/nymtech/nym/pull/1520
[#1573]: https://github.com/nymtech/nym/pull/1573
[#1576]: https://github.com/nymtech/nym/pull/1576
## [v1.0.1](https://github.com/nymtech/nym/tree/v1.0.1) (2022-05-04)
Cargo.lock
Generated
+1
View File
@@ -5211,6 +5211,7 @@ version = "0.1.0"
dependencies = [
"nymsphinx-addressing",
"ordered-buffer",
"thiserror",
]
[[package]]
clients/client-core/src/client/real_messages_control/acknowledgement_control/input_message_listener.rs
-1
View File
@@ -133,7 +133,6 @@ where
let prepared_fragment = self
.message_preparer
.prepare_chunk_for_sending(chunk_clone, topology, &self.ack_key, &recipient)
.await
.unwrap();
real_messages.push(RealMessage::new(
clients/client-core/src/client/real_messages_control/acknowledgement_control/retransmission_request_listener.rs
-1
View File
@@ -83,7 +83,6 @@ where
let prepared_fragment = self
.message_preparer
.prepare_chunk_for_sending(chunk_clone, topology_ref, &self.ack_key, packet_recipient)
.await
.unwrap();
// if we have the ONLY strong reference to the ack data, it means it was removed from the
clients/socks5/src/socks/mixnet_responses.rs
+7
View File
@@ -54,6 +54,13 @@ impl MixnetResponseListener {
return;
}
Ok(Message::Response(data)) => data,
Ok(Message::NetworkRequesterResponse(r)) => {
error!(
"Network requester failed on connection id {} with error: {}",
r.connection_id, r.network_requester_error
);
return;
}
};
self.controller_sender
clients/webassembly/src/client/mod.rs
-1
View File
@@ -197,7 +197,6 @@ impl NymClient {
// 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)
.await
.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);
common/cosmwasm-smart-contracts/mixnet-contract/src/msg.rs
+1 -1
View File
@@ -217,7 +217,7 @@ pub enum QueryMsg {
#[serde(rename_all = "snake_case")]
pub struct MigrateMsg {
pub mixnet_denom: String,
nodes_to_remove: Option<Vec<NodeToRemove>>,
pub nodes_to_remove: Option<Vec<NodeToRemove>>,
}
impl MigrateMsg {
common/nymsphinx/src/preparer/mod.rs
+4 -6
View File
@@ -213,7 +213,7 @@ where
/// - compute vk_b = g^x || v_b
/// - compute sphinx_plaintext = SURB_ACK || g^x || v_b
/// - compute sphinx_packet = Sphinx(recipient, sphinx_plaintext)
pub async fn prepare_chunk_for_sending(
pub fn prepare_chunk_for_sending(
&mut self,
fragment: Fragment,
topology: &NymTopology,
@@ -222,8 +222,7 @@ where
) -> Result<PreparedFragment, NymTopologyError> {
// create an ack
let (ack_delay, surb_ack_bytes) = self
.generate_surb_ack(fragment.fragment_identifier(), topology, ack_key)
.await?
.generate_surb_ack(fragment.fragment_identifier(), topology, ack_key)?
.prepare_for_sending();
// TODO:
@@ -294,7 +293,7 @@ where
}
/// Construct an acknowledgement SURB for the given [`FragmentIdentifier`]
async fn generate_surb_ack(
fn generate_surb_ack(
&mut self,
fragment_id: FragmentIdentifier,
topology: &NymTopology,
@@ -357,8 +356,7 @@ where
// gateways could not distinguish reply packets from normal messages due to lack of said acks
// note: the ack delay is irrelevant since we do not know the delay of actual surb
let (_, surb_ack_bytes) = self
.generate_surb_ack(reply_id, topology, ack_key)
.await?
.generate_surb_ack(reply_id, topology, ack_key)?
.prepare_for_sending();
let zero_pad_len = self.packet_size.plaintext_size()
common/socks5/requests/Cargo.toml
+1
View File
@@ -9,3 +9,4 @@ edition = "2021"
[dependencies]
nymsphinx-addressing = { path = "../../../common/nymsphinx/addressing" }
ordered-buffer = {path = "../ordered-buffer"}
thiserror = "1"
common/socks5/requests/src/lib.rs
+5
View File
@@ -1,7 +1,12 @@
// Copyright 2020-2022 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
pub mod msg;
pub mod network_requester_response;
pub mod request;
pub mod response;
pub use msg::*;
pub use network_requester_response::*;
pub use request::*;
pub use response::*;
common/socks5/requests/src/msg.rs
+28 -15
View File
@@ -1,36 +1,40 @@
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
// Copyright 2020-2022 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use thiserror::Error;
use crate::network_requester_response::{Error as NrError, NetworkRequesterResponse};
use crate::request::{Request, RequestError};
use crate::response::{Response, ResponseError};
#[derive(Debug)]
#[derive(Debug, Error)]
pub enum MessageError {
#[error("{0}")]
Request(RequestError),
Response(ResponseError),
NoData,
UnknownMessageType,
}
impl std::fmt::Display for MessageError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
MessageError::Request(r) => write!(f, "{}", r),
MessageError::Response(r) => write!(f, "{:?}", r),
MessageError::NoData => write!(f, "no data provided"),
MessageError::UnknownMessageType => write!(f, "unknown message type received"),
}
}
#[error("{0:?}")]
Response(ResponseError),
#[error("{0}")]
NetworkRequesterResponseError(NrError),
#[error("no data")]
NoData,
#[error("unknown message type received")]
UnknownMessageType,
}
pub enum Message {
Request(Request),
Response(Response),
NetworkRequesterResponse(NetworkRequesterResponse),
}
impl Message {
const REQUEST_FLAG: u8 = 0;
const RESPONSE_FLAG: u8 = 1;
const NR_RESPONSE_FLAG: u8 = 2;
pub fn conn_id(&self) -> u64 {
match self {
@@ -39,6 +43,7 @@ impl Message {
Request::Send(conn_id, _, _) => *conn_id,
},
Message::Response(resp) => resp.connection_id,
Message::NetworkRequesterResponse(resp) => resp.connection_id,
}
}
@@ -49,6 +54,7 @@ impl Message {
Request::Send(_, data, _) => data.len(),
},
Message::Response(resp) => resp.data.len(),
Message::NetworkRequesterResponse(_) => 0,
}
}
@@ -65,6 +71,10 @@ impl Message {
Response::try_from_bytes(&b[1..])
.map(Message::Response)
.map_err(MessageError::Response)
} else if b[0] == Self::NR_RESPONSE_FLAG {
NetworkRequesterResponse::try_from_bytes(&b[1..])
.map(Message::NetworkRequesterResponse)
.map_err(MessageError::NetworkRequesterResponseError)
} else {
Err(MessageError::UnknownMessageType)
}
@@ -78,6 +88,9 @@ impl Message {
Self::Response(r) => std::iter::once(Self::RESPONSE_FLAG)
.chain(r.into_bytes().iter().cloned())
.collect(),
Self::NetworkRequesterResponse(r) => std::iter::once(Self::NR_RESPONSE_FLAG)
.chain(r.into_bytes().iter().cloned())
.collect(),
}
}
}
common/socks5/requests/src/network_requester_response.rs
+112
View File
@@ -0,0 +1,112 @@
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::ConnectionId;
#[derive(Debug)]
pub struct NetworkRequesterResponse {
pub connection_id: ConnectionId,
pub network_requester_error: String,
}
#[derive(Debug, thiserror::Error, PartialEq, Eq)]
pub enum Error {
#[error("no data provided")]
NoData,
#[error("not enough bytes to recover the connection id")]
ConnectionIdTooShort,
#[error("message is not utf8 encoded")]
MalformedErrorMessage(#[from] std::string::FromUtf8Error),
}
impl NetworkRequesterResponse {
pub fn new(connection_id: ConnectionId, network_requester_error: String) -> Self {
NetworkRequesterResponse {
connection_id,
network_requester_error,
}
}
pub fn try_from_bytes(b: &[u8]) -> Result<NetworkRequesterResponse, Error> {
if b.is_empty() {
return Err(Error::NoData);
}
if b.len() < 8 {
return Err(Error::ConnectionIdTooShort);
}
let mut connection_id_bytes = b.to_vec();
let network_requester_error_bytes = connection_id_bytes.split_off(8);
let connection_id = u64::from_be_bytes([
connection_id_bytes[0],
connection_id_bytes[1],
connection_id_bytes[2],
connection_id_bytes[3],
connection_id_bytes[4],
connection_id_bytes[5],
connection_id_bytes[6],
connection_id_bytes[7],
]);
let network_requester_error = String::from_utf8(network_requester_error_bytes)?;
Ok(NetworkRequesterResponse {
connection_id,
network_requester_error,
})
}
pub fn into_bytes(self) -> Vec<u8> {
self.connection_id
.to_be_bytes()
.iter()
.cloned()
.chain(self.network_requester_error.into_bytes().into_iter())
.collect()
}
}
#[cfg(test)]
mod network_requester_response_serde_tests {
use super::*;
#[test]
fn simple_serde() {
let conn_id = 42;
let network_requester_error = String::from("This is a test msg");
let response = NetworkRequesterResponse::new(conn_id, network_requester_error.clone());
let bytes = response.into_bytes();
let deserialized_response = NetworkRequesterResponse::try_from_bytes(&bytes).unwrap();
assert_eq!(conn_id, deserialized_response.connection_id);
assert_eq!(
network_requester_error,
deserialized_response.network_requester_error
);
}
#[test]
fn deserialization_errors() {
let err = NetworkRequesterResponse::try_from_bytes(&[]).err().unwrap();
assert_eq!(err, Error::NoData);
let bytes: [u8; 5] = [1, 2, 3, 4, 5];
let err = NetworkRequesterResponse::try_from_bytes(&bytes)
.err()
.unwrap();
assert_eq!(err, Error::ConnectionIdTooShort);
let bytes: Vec<u8> = 42u64
.to_be_bytes()
.into_iter()
.chain([0, 159, 146, 150].into_iter())
.collect();
let err = NetworkRequesterResponse::try_from_bytes(&bytes)
.err()
.unwrap();
assert!(matches!(err, Error::MalformedErrorMessage(_)));
}
}
common/socks5/requests/src/request.rs
+18 -24
View File
@@ -1,6 +1,9 @@
// Copyright 2020-2022 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use nymsphinx_addressing::clients::{Recipient, RecipientFormattingError};
use std::convert::TryFrom;
use std::fmt::{self};
use thiserror::Error;
pub type ConnectionId = u64;
pub type RemoteAddress = String;
@@ -12,39 +15,30 @@ pub enum RequestFlag {
Send = 1,
}
#[derive(Debug)]
#[derive(Debug, Error)]
pub enum RequestError {
#[error("not enough bytes to recover the length of the address")]
AddressLengthTooShort,
#[error("not enough bytes to recover the address")]
AddressTooShort,
#[error("not enough bytes to recover the connection id")]
ConnectionIdTooShort,
#[error("no data provided")]
NoData,
#[error("request of unknown type")]
UnknownRequestFlag,
#[error("too short return address")]
ReturnAddressTooShort,
#[error("malformed return address - {0}")]
MalformedReturnAddress(RecipientFormattingError),
}
impl fmt::Display for RequestError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> fmt::Result {
match self {
RequestError::AddressLengthTooShort => {
write!(f, "not enough bytes to recover the length of the address")
}
RequestError::AddressTooShort => write!(f, "not enough bytes to recover the address"),
RequestError::ConnectionIdTooShort => {
write!(f, "not enough bytes to recover the connection id")
}
RequestError::NoData => write!(f, "no data provided"),
RequestError::UnknownRequestFlag => write!(f, "request of unknown type"),
RequestError::ReturnAddressTooShort => write!(f, "too short return address"),
RequestError::MalformedReturnAddress(recipient_err) => {
write!(f, "malformed return address - {}", recipient_err)
}
}
}
}
impl std::error::Error for RequestError {}
impl RequestError {
pub fn is_malformed_return(&self) -> bool {
matches!(self, RequestError::MalformedReturnAddress(_))
common/socks5/requests/src/response.rs
+8 -1
View File
@@ -1,8 +1,15 @@
// Copyright 2020-2022 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use thiserror::Error;
use crate::ConnectionId;
#[derive(Debug, PartialEq, Eq)]
#[derive(Debug, Error, PartialEq, Eq)]
pub enum ResponseError {
#[error("not enough bytes to recover the connection id")]
ConnectionIdTooShort,
#[error("no data provided")]
NoData,
}
/// A remote network response retrieved by the Socks5 service provider. This
common/task/src/shutdown.rs
+11 -2
View File
@@ -5,13 +5,14 @@ use std::time::Duration;
use tokio::sync::watch::{self, error::SendError};
const SHUTDOWN_TIMER_SECS: u64 = 5;
const DEFAULT_SHUTDOWN_TIMER_SECS: u64 = 5;
/// Used to notify other tasks to gracefully shutdown
#[derive(Debug)]
pub struct ShutdownNotifier {
notify_tx: watch::Sender<()>,
notify_rx: Option<watch::Receiver<()>>,
shutdown_timer_secs: u64,
}
impl Default for ShutdownNotifier {
@@ -20,11 +21,19 @@ impl Default for ShutdownNotifier {
Self {
notify_tx,
notify_rx: Some(notify_rx),
shutdown_timer_secs: DEFAULT_SHUTDOWN_TIMER_SECS,
}
}
}
impl ShutdownNotifier {
pub fn new(shutdown_timer_secs: u64) -> Self {
Self {
shutdown_timer_secs,
..Default::default()
}
}
pub fn subscribe(&self) -> ShutdownListener {
ShutdownListener::new(
self.notify_rx
@@ -50,7 +59,7 @@ impl ShutdownNotifier {
_ = tokio::signal::ctrl_c() => {
log::info!("Forcing shutdown");
}
_ = tokio::time::sleep(Duration::from_secs(SHUTDOWN_TIMER_SECS)) => {
_ = tokio::time::sleep(Duration::from_secs(self.shutdown_timer_secs)) => {
log::info!("Timout reached, forcing shutdown");
},
}
contracts/CHANGELOG.md
+6
View File
@@ -4,12 +4,18 @@
- vesting-contract: added queries for delegation timestamps and paged query for all vesting delegations in the contract ([#1569])
### Changed
- mixnet-contract: compounding delegator rewards now happens instantaneously as opposed to having to wait for the current epoch to finish ([#1571])
### Fixed
- vesting-contract: the contract now correctly stores delegations with their timestamp as opposed to using block height ([#1544])
- mixnet-contract: compounding delegator rewards is now possible even if the associated mixnode had already unbonded ([#1571])
[#1544]: https://github.com/nymtech/nym/pull/1544
[#1569]: https://github.com/nymtech/nym/pull/1569
[#1569]: https://github.com/nymtech/nym/pull/1571
## [nym-contracts-v1.0.1](https://github.com/nymtech/nym/tree/nym-contracts-v1.0.1) (2022-06-22)
contracts/mixnet/src/rewards/transactions.rs
+24 -23
View File
@@ -8,7 +8,6 @@ use super::storage::{
use crate::constants;
use crate::contract::debug_with_visibility;
use crate::delegations::storage as delegations_storage;
use crate::delegations::transactions::_try_delegate_to_mixnode;
use crate::error::ContractError;
use crate::mixnet_contract_settings::storage::mix_denom;
use crate::mixnodes::storage::mixnodes;
@@ -18,7 +17,7 @@ use crate::rewards::helpers;
use crate::support::helpers::{is_authorized, operator_cost_at_epoch};
use cosmwasm_std::{
coins, wasm_execute, Addr, Api, BankMsg, Coin, DepsMut, Env, MessageInfo, Order, Response,
Storage, Uint128,
StdResult, Storage, Uint128,
};
use cw_storage_plus::Bound;
use mixnet_contract_common::events::{
@@ -450,18 +449,15 @@ pub fn try_compound_delegator_reward(
pub fn _try_compound_delegator_reward(
block_height: u64,
mut deps: DepsMut<'_>,
deps: DepsMut<'_>,
owner_address: &str,
mix_identity: &str,
proxy: Option<Addr>,
) -> Result<Uint128, ContractError> {
let delegation_map = crate::delegations::storage::delegations();
let mix_denom = mix_denom(deps.storage)?;
let key = mixnet_contract_common::delegation::generate_storage_key(
&deps.api.addr_validate(owner_address)?,
proxy.as_ref(),
);
let owner = deps.api.addr_validate(owner_address)?;
let key = mixnet_contract_common::delegation::generate_storage_key(&owner, proxy.as_ref());
let reward = calculate_delegator_reward(deps.storage, deps.api, key.clone(), mix_identity)?;
let mut total_delegation_delegate = Uint128::zero();
@@ -469,8 +465,7 @@ pub fn _try_compound_delegator_reward(
let delegation_heights = delegation_map
.prefix((mix_identity.to_string(), key.clone()))
.keys(deps.storage, None, None, cosmwasm_std::Order::Ascending)
.filter_map(|v| v.ok())
.collect::<Vec<u64>>();
.collect::<StdResult<Vec<_>>>()?;
for h in delegation_heights {
let delegation =
@@ -494,30 +489,36 @@ pub fn _try_compound_delegator_reward(
// since we know that the target node exists and because the total_delegation bucket
// entry is created whenever the node itself is added, the unwrap here is fine
// as the entry MUST exist
Ok(total_delegation
.unwrap()
.saturating_sub(total_delegation_delegate))
Ok(total_delegation.unwrap() + reward)
},
)?;
_try_delegate_to_mixnode(
deps.branch(),
block_height,
mix_identity,
owner_address,
// let's simplify the entire procedure. Rather than creating a fresh delegation on the mixnode
// via `_try_delegate_to_mixnode` and then waiting for reconcile to happen,
// just save it directly to the storage right now.
// my reasoning for that is simple: `_try_delegate_to_mixnode` could fail if the node the
// delegator has delegated to no longer exists.
let delegation = Delegation::new(
owner,
mix_identity.into(),
Coin {
amount: compounded_delegation,
denom: mix_denom,
},
block_height,
proxy,
);
delegation_map.save(
deps.storage,
(mix_identity.into(), key.clone(), block_height),
&delegation,
)?;
}
{
if let Some(mut bond) = mixnodes().may_load(deps.storage, mix_identity)? {
bond.accumulated_rewards = Some(bond.accumulated_rewards().saturating_sub(reward));
mixnodes().save(deps.storage, mix_identity, &bond, block_height)?;
}
if let Some(mut bond) = mixnodes().may_load(deps.storage, mix_identity)? {
bond.accumulated_rewards = Some(bond.accumulated_rewards().saturating_sub(reward));
mixnodes().save(deps.storage, mix_identity, &bond, block_height)?;
}
DELEGATOR_REWARD_CLAIMED_HEIGHT.save(
service-providers/network-requester/src/core.rs
+14 -3
View File
@@ -13,7 +13,9 @@ use log::*;
use nymsphinx::addressing::clients::Recipient;
use nymsphinx::receiver::ReconstructedMessage;
use proxy_helpers::connection_controller::{Controller, ControllerCommand, ControllerSender};
use socks5_requests::{ConnectionId, Message as Socks5Message, Request, Response};
use socks5_requests::{
ConnectionId, Message as Socks5Message, NetworkRequesterResponse, Request, Response,
};
use statistics_common::collector::StatisticsSender;
use std::path::PathBuf;
use std::sync::atomic::{AtomicUsize, Ordering};
@@ -192,7 +194,16 @@ impl ServiceProvider {
return_address: Recipient,
) {
if !self.open_proxy && !self.outbound_request_filter.check(&remote_addr) {
log::info!("Domain {:?} failed filter check", remote_addr);
let log_msg = format!("Domain {:?} failed filter check", remote_addr);
log::info!("{}", log_msg);
mix_input_sender
.unbounded_send((
Socks5Message::NetworkRequesterResponse(NetworkRequesterResponse::new(
conn_id, log_msg,
)),
return_address,
))
.unwrap();
return;
}
@@ -275,7 +286,7 @@ impl ServiceProvider {
self.handle_proxy_send(controller_sender, conn_id, data, closed)
}
},
Socks5Message::Response(_) => {}
Socks5Message::Response(_) | Socks5Message::NetworkRequesterResponse(_) => {}
}
}
validator-api/src/contract_cache/mod.rs
+15 -7
View File
@@ -281,13 +281,21 @@ impl<C> ValidatorCacheRefresher<C> {
while !shutdown.is_shutdown() {
tokio::select! {
_ = interval.tick() => {
if let Err(err) = self.refresh_cache().await {
error!("Failed to refresh validator cache - {}", err);
} else {
// relaxed memory ordering is fine here. worst case scenario network monitor
// will just have to wait for an additional backoff to see the change.
// And so this will not really incur any performance penalties by setting it every loop iteration
self.cache.initialised.store(true, Ordering::Relaxed)
tokio::select! {
biased;
_ = shutdown.recv() => {
trace!("ValidatorCacheRefresher: Received shutdown");
}
ret = self.refresh_cache() => {
if let Err(err) = ret {
error!("Failed to refresh validator cache - {}", err);
} else {
// relaxed memory ordering is fine here. worst case scenario network monitor
// will just have to wait for an additional backoff to see the change.
// And so this will not really incur any performance penalties by setting it every loop iteration
self.cache.initialised.store(true, Ordering::Relaxed)
}
}
}
}
_ = shutdown.recv() => {
validator-api/src/main.rs
+4 -2
View File
@@ -568,7 +568,8 @@ async fn run_validator_api(matches: ArgMatches<'static>) -> Result<()> {
let signing_nymd_client = Client::new_signing(&config);
let liftoff_notify = Arc::new(Notify::new());
let shutdown = ShutdownNotifier::default();
// We need a bigger timeout
let shutdown = ShutdownNotifier::new(10);
// let's build our rocket!
let rocket = setup_rocket(
@@ -609,7 +610,8 @@ async fn run_validator_api(matches: ArgMatches<'static>) -> Result<()> {
// spawn rewarded set updater
let mut rewarded_set_updater =
RewardedSetUpdater::new(signing_nymd_client, validator_cache.clone(), storage).await?;
tokio::spawn(async move { rewarded_set_updater.run().await.unwrap() });
let shutdown_listener = shutdown.subscribe();
tokio::spawn(async move { rewarded_set_updater.run(shutdown_listener).await.unwrap() });
validator_cache_listener
} else {
validator-api/src/network_monitor/chunker.rs
+4 -4
View File
@@ -12,6 +12,7 @@ use topology::NymTopology;
const DEFAULT_AVERAGE_PACKET_DELAY: Duration = Duration::from_millis(200);
const DEFAULT_AVERAGE_ACK_DELAY: Duration = Duration::from_millis(200);
#[derive(Clone)]
pub(crate) struct Chunker {
rng: OsRng,
message_preparer: MessagePreparer<OsRng>,
@@ -30,7 +31,7 @@ impl Chunker {
}
}
pub(crate) async fn prepare_packets_from(
pub(crate) fn prepare_packets_from(
&mut self,
message: Vec<u8>,
topology: &NymTopology,
@@ -40,10 +41,10 @@ impl Chunker {
// but without some significant API changes in the `MessagePreparer` this was the easiest
// way to being able to have variable sender address.
self.message_preparer.set_sender_address(packet_sender);
self.prepare_packets(message, topology, packet_sender).await
self.prepare_packets(message, topology, packet_sender)
}
async fn prepare_packets(
fn prepare_packets(
&mut self,
message: Vec<u8>,
topology: &NymTopology,
@@ -62,7 +63,6 @@ impl Chunker {
let prepared_fragment = self
.message_preparer
.prepare_chunk_for_sending(message_chunk, topology, &ack_key, &packet_sender)
.await
.unwrap();
mix_packets.push(prepared_fragment.mix_packet);
validator-api/src/network_monitor/monitor/gateways_pinger.rs
+17 -4
View File
@@ -7,6 +7,7 @@ use crypto::asymmetric::identity;
use crypto::asymmetric::identity::PUBLIC_KEY_LENGTH;
use log::{debug, info, trace, warn};
use std::time::Duration;
use task::ShutdownListener;
use tokio::time::{sleep, Instant};
// TODO: should it perhaps be moved to config along other timeout values?
@@ -143,10 +144,22 @@ impl GatewayPinger {
info!("Pinging all active gateways took {:?}", time_taken);
}
pub(crate) async fn run(&self) {
loop {
sleep(self.pinging_interval).await;
self.ping_and_cleanup_all_gateways().await
pub(crate) async fn run(&self, mut shutdown: ShutdownListener) {
while !shutdown.is_shutdown() {
tokio::select! {
_ = sleep(self.pinging_interval) => {
tokio::select! {
biased;
_ = shutdown.recv() => {
trace!("GatewaysPinger: Received shutdown");
}
_ = self.ping_and_cleanup_all_gateways() => (),
}
}
_ = shutdown.recv() => {
trace!("GatewaysPinger: Received shutdown");
}
}
}
}
}
validator-api/src/network_monitor/monitor/mod.rs
+19 -7
View File
@@ -122,11 +122,15 @@ impl Monitor {
let mut packets = Vec::with_capacity(routes.len());
for route in routes {
packets.push(
self.packet_preparer
.prepare_test_route_viability_packets(route, self.route_test_packets)
.await,
);
let mut packet_preparer = self.packet_preparer.clone();
let route = route.clone();
let route_test_packets = self.route_test_packets;
let gateway_packets = tokio::spawn(async move {
packet_preparer.prepare_test_route_viability_packets(&route, route_test_packets)
})
.await
.unwrap();
packets.push(gateway_packets);
}
self.received_processor.set_route_test_nonce().await;
@@ -306,12 +310,20 @@ impl Monitor {
.await;
self.packet_sender
.spawn_gateways_pinger(self.gateway_ping_interval);
.spawn_gateways_pinger(self.gateway_ping_interval, shutdown.clone());
let mut run_interval = tokio::time::interval(self.run_interval);
while !shutdown.is_shutdown() {
tokio::select! {
_ = run_interval.tick() => self.test_run().await,
_ = run_interval.tick() => {
tokio::select! {
biased;
_ = shutdown.recv() => {
trace!("UpdateHandler: Received shutdown");
}
_ = self.test_run() => (),
}
}
_ = shutdown.recv() => {
trace!("UpdateHandler: Received shutdown");
}
validator-api/src/network_monitor/monitor/preparer.rs
+11 -17
View File
@@ -117,6 +117,7 @@ pub(crate) struct PreparedPackets {
pub(super) invalid_gateways: Vec<InvalidNode>,
}
#[derive(Clone)]
pub(crate) struct PacketPreparer {
system_version: String,
chunker: Option<Chunker>,
@@ -151,7 +152,7 @@ impl PacketPreparer {
}
}
async fn wrap_test_packet(
fn wrap_test_packet(
&mut self,
packet: &TestPacket,
topology: &NymTopology,
@@ -162,12 +163,11 @@ impl PacketPreparer {
if self.chunker.is_none() {
self.chunker = Some(Chunker::new(packet_recipient));
}
let mut mix_packets = self
.chunker
.as_mut()
.unwrap()
.prepare_packets_from(packet.to_bytes(), topology, packet_recipient)
.await;
let mut mix_packets = self.chunker.as_mut().unwrap().prepare_packets_from(
packet.to_bytes(),
topology,
packet_recipient,
);
assert_eq!(
mix_packets.len(),
1,
@@ -351,7 +351,7 @@ impl PacketPreparer {
)
}
pub(crate) async fn prepare_test_route_viability_packets(
pub(crate) fn prepare_test_route_viability_packets(
&mut self,
route: &TestRoute,
num: usize,
@@ -360,9 +360,7 @@ impl PacketPreparer {
let test_packet = route.self_test_packet();
let recipient = self.create_packet_sender(route.gateway());
for _ in 0..num {
let mix_packet = self
.wrap_test_packet(&test_packet, route.topology(), recipient)
.await;
let mix_packet = self.wrap_test_packet(&test_packet, route.topology(), recipient);
mix_packets.push(mix_packet)
}
@@ -451,9 +449,7 @@ impl PacketPreparer {
let topology = test_route.substitute_mix(mixnode);
// produce n mix packets
for _ in 0..self.per_node_test_packets {
let mix_packet = self
.wrap_test_packet(&test_packet, &topology, recipient)
.await;
let mix_packet = self.wrap_test_packet(&test_packet, &topology, recipient);
mix_packets.push(mix_packet);
}
}
@@ -476,9 +472,7 @@ impl PacketPreparer {
let topology = test_route.substitute_gateway(gateway);
// produce n mix packets
for _ in 0..self.per_node_test_packets {
let mix_packet = self
.wrap_test_packet(&test_packet, &topology, recipient)
.await;
let mix_packet = self.wrap_test_packet(&test_packet, &topology, recipient);
gateway_mix_packets.push(mix_packet);
}
validator-api/src/network_monitor/monitor/processor.rs
+17 -13
View File
@@ -140,8 +140,7 @@ impl ReceivedProcessor {
self.permit_changer = Some(permit_sender);
tokio::spawn(async move {
loop {
let permit = wait_for_permit(&mut permit_receiver, &*inner).await;
while let Some(permit) = wait_for_permit(&mut permit_receiver, &*inner).await {
receive_or_release_permit(&mut permit_receiver, permit).await;
}
@@ -151,16 +150,20 @@ impl ReceivedProcessor {
) {
loop {
tokio::select! {
permit_receiver = permit_receiver.next() => match permit_receiver.unwrap() {
LockPermit::Release => return,
LockPermit::Free => error!("somehow we got notification that the lock is free to take while we already hold it!"),
permit_receiver = permit_receiver.next() => match permit_receiver {
Some(LockPermit::Release) => return,
Some(LockPermit::Free) => error!("somehow we got notification that the lock is free to take while we already hold it!"),
None => return,
},
messages = inner.packets_receiver.next() => {
for message in messages.expect("packet receiver has died!") {
if let Err(err) = inner.on_message(message) {
warn!(target: "Monitor", "failed to process received gateway message - {}", err)
messages = inner.packets_receiver.next() => match messages {
Some(messages) => {
for message in messages {
if let Err(err) = inner.on_message(message) {
warn!(target: "Monitor", "failed to process received gateway message - {}", err)
}
}
}
None => return,
},
}
}
@@ -172,14 +175,15 @@ impl ReceivedProcessor {
async fn wait_for_permit<'a>(
permit_receiver: &mut mpsc::Receiver<LockPermit>,
inner: &'a Mutex<ReceivedProcessorInner>,
) -> MutexGuard<'a, ReceivedProcessorInner> {
) -> Option<MutexGuard<'a, ReceivedProcessorInner>> {
loop {
match permit_receiver.next().await.unwrap() {
match permit_receiver.next().await {
// we should only ever get this on the very first run
LockPermit::Release => debug!(
Some(LockPermit::Release) => debug!(
"somehow got request to drop our lock permit while we do not hold it!"
),
LockPermit::Free => return inner.lock().await,
Some(LockPermit::Free) => return Some(inner.lock().await),
None => return None,
}
}
}
validator-api/src/network_monitor/monitor/receiver.rs
+4 -3
View File
@@ -59,6 +59,10 @@ impl PacketReceiver {
pub(crate) async fn run(&mut self, mut shutdown: ShutdownListener) {
while !shutdown.is_shutdown() {
tokio::select! {
biased;
_ = shutdown.recv() => {
trace!("UpdateHandler: Received shutdown");
}
// unwrap here is fine as it can only return a `None` if the PacketSender has died
// and if that was the case, then the entire monitor is already in an undefined state
update = self.clients_updater.next() => self.process_gateway_update(update.unwrap()),
@@ -68,9 +72,6 @@ impl PacketReceiver {
Some((_gateway_id, message)) = self.gateways_reader.stream_map().next() => {
self.process_gateway_messages(message)
}
_ = shutdown.recv() => {
trace!("UpdateHandler: Received shutdown");
}
}
}
}
validator-api/src/network_monitor/monitor/sender.rs
+7 -2
View File
@@ -24,6 +24,7 @@ use std::pin::Pin;
use std::sync::Arc;
use std::task::Poll;
use std::time::Duration;
use task::ShutdownListener;
use gateway_client::bandwidth::BandwidthController;
@@ -176,7 +177,11 @@ impl PacketSender {
}
}
pub(crate) fn spawn_gateways_pinger(&self, pinging_interval: Duration) {
pub(crate) fn spawn_gateways_pinger(
&self,
pinging_interval: Duration,
shutdown: ShutdownListener,
) {
let gateway_pinger = GatewayPinger::new(
self.active_gateway_clients.clone(),
self.fresh_gateway_client_data
@@ -185,7 +190,7 @@ impl PacketSender {
pinging_interval,
);
tokio::spawn(async move { gateway_pinger.run().await });
tokio::spawn(async move { gateway_pinger.run(shutdown).await });
}
fn new_gateway_client_handle(
validator-api/src/node_status_api/cache.rs
+16 -5
View File
@@ -123,17 +123,28 @@ impl NodeStatusCacheRefresher {
let mut fallback_interval = time::interval(self.fallback_caching_interval);
while !shutdown.is_shutdown() {
tokio::select! {
biased;
_ = shutdown.recv() => {
log::trace!("NodeStatusCacheRefresher: Received shutdown");
}
// Update node status cache when the contract cache / validator cache is updated
Ok(_) = self.contract_cache_listener.changed() => {
self.update_on_notify(&mut fallback_interval).await;
tokio::select! {
_ = self.update_on_notify(&mut fallback_interval) => (),
_ = shutdown.recv() => {
log::trace!("NodeStatusCacheRefresher: Received shutdown");
}
}
}
// ... however, if we don't receive any notifications we fall back to periodic
// refreshes
_ = fallback_interval.tick() => {
self.update_on_timer().await;
}
_ = shutdown.recv() => {
log::trace!("NodeStatusCacheRefresher: Received shutdown");
tokio::select! {
_ = self.update_on_timer() => (),
_ = shutdown.recv() => {
log::trace!("NodeStatusCacheRefresher: Received shutdown");
}
}
}
}
}
validator-api/src/node_status_api/uptime_updater.rs
+14 -8
View File
@@ -72,14 +72,20 @@ impl HistoricalUptimeUpdater {
while !shutdown.is_shutdown() {
tokio::select! {
_ = sleep(ONE_DAY) => {
if let Err(err) = self.update_uptimes().await {
// normally that would have been a warning rather than an error,
// however, in this case it implies some underlying issues with our database
// that might affect the entire program
error!(
"We failed to update daily uptimes of active nodes - {}",
err
)
tokio::select! {
biased;
_ = shutdown.recv() => {
trace!("UpdateHandler: Received shutdown");
}
Err(err) = self.update_uptimes() => {
// normally that would have been a warning rather than an error,
// however, in this case it implies some underlying issues with our database
// that might affect the entire program
error!(
"We failed to update daily uptimes of active nodes - {}",
err
);
}
}
}
_ = shutdown.recv() => {
validator-api/src/rewarded_set_updater/mod.rs
+16 -3
View File
@@ -24,6 +24,7 @@ use rand::rngs::OsRng;
use std::collections::HashSet;
use std::ops::Deref;
use std::time::Duration;
use task::ShutdownListener;
use time::OffsetDateTime;
use tokio::time::sleep;
use validator_client::nymd::{Coin, SigningNymdClient};
@@ -328,11 +329,14 @@ impl RewardedSetUpdater {
Ok(())
}
pub(crate) async fn run(&mut self) -> Result<(), RewardingError> {
pub(crate) async fn run(
&mut self,
mut shutdown: ShutdownListener,
) -> Result<(), RewardingError> {
self.validator_cache.wait_for_initial_values().await;
let mut epoch = self.epoch().await?;
loop {
while !shutdown.is_shutdown() {
// wait until the cache refresher determined its time to update the rewarded/active sets
let time = OffsetDateTime::now_utc().unix_timestamp();
epoch.update_to_latest(self).await?;
@@ -351,7 +355,16 @@ impl RewardedSetUpdater {
);
// Sleep at most 300 before checking again, to keep logs busy
let s = time_to_epoch_change.min(300).max(0) as u64;
sleep(Duration::from_secs(s)).await;
tokio::select! {
_ = sleep(Duration::from_secs(s)) => {
trace!("Checking again for updating rewarded/active sets");
}
_ = shutdown.recv() => {
trace!("RewardedSetUpdater: Received shutdown");
// This break should not be necessary, but there's a following sleep after this
break;
}
}
}
// allow some blocks to pass
sleep(Duration::from_secs(10)).await;