Pending endpoints (#1150)

* Add QueryPendingDelegationEvents QueryMsg

* Add pending_delegation_events to nymd client

* Add pending delegation events to wallet

* Get rid of double epoch accounting

* Fix reward saving

* Try batching operations to reduce fees

* Bundle all transactions into one

* make nice

Author: durch

common/client-libs/validator-client/src/client.rs

@@ -3,6 +3,7 @@
 
 use crate::{validator_api, ValidatorClientError};
 use coconut_interface::{BlindSignRequestBody, BlindedSignatureResponse, VerificationKeyResponse};
+use mixnet_contract_common::mixnode::DelegationEvent;
 use mixnet_contract_common::{GatewayBond, IdentityKeyRef, Interval, MixNodeBond};
 use network_defaults::DEFAULT_NETWORK;
 use url::Url;
@@ -270,7 +271,20 @@ impl<C> Client<C> {
             .u128())
     }
 
-    pub async fn get_current_epoch(&self) -> Result<Option<Interval>, ValidatorClientError>
+    pub async fn get_pending_delegation_events(
+        &self,
+        owner_address: String,
+    ) -> Result<Vec<DelegationEvent>, ValidatorClientError>
+    where
+        C: CosmWasmClient + Sync,
+    {
+        Ok(self
+            .nymd
+            .get_pending_delegation_events(owner_address)
+            .await?)
+    }
+
+    pub async fn get_current_epoch(&self) -> Result<Interval, ValidatorClientError>
     where
         C: CosmWasmClient + Sync,
     {

common/client-libs/validator-client/src/nymd/mod.rs

@@ -14,6 +14,7 @@ use cosmrs::rpc::HttpClientUrl;
 use cosmwasm_std::{Coin, Uint128};
 pub use fee::gas_price::GasPrice;
 use fee::helpers::Operation;
+use mixnet_contract_common::mixnode::DelegationEvent;
 use mixnet_contract_common::{
     ContractStateParams, Delegation, ExecuteMsg, Gateway, GatewayBond, GatewayOwnershipResponse,
     IdentityKey, Interval, LayerDistribution, MixNode, MixNodeBond, MixOwnershipResponse,
@@ -317,7 +318,20 @@ impl<C> NymdClient<C> {
             .await
     }
 
-    pub async fn get_current_epoch(&self) -> Result<Option<Interval>, NymdError>
+    pub async fn get_pending_delegation_events(
+        &self,
+        owner_address: String,
+    ) -> Result<Vec<DelegationEvent>, NymdError>
+    where
+        C: CosmWasmClient + Sync,
+    {
+        let request = QueryMsg::GetPendingDelegationEvents { owner_address };
+        self.client
+            .query_contract_smart(self.mixnet_contract_address()?, &request)
+            .await
+    }
+
+    pub async fn get_current_epoch(&self) -> Result<Interval, NymdError>
     where
         C: CosmWasmClient + Sync,
     {

common/cosmwasm-smart-contracts/mixnet-contract/src/msg.rs

@@ -176,6 +176,9 @@ pub enum QueryMsg {
         address: String,
         mix_identity: IdentityKey,
     },
+    GetPendingDelegationEvents {
+        owner_address: String,
+    },
 }
 
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]

contracts/mixnet/src/contract.rs

@@ -5,7 +5,9 @@ use crate::constants::{ACTIVE_SET_WORK_FACTOR, INTERVAL_REWARD_PERCENT, SYBIL_RE
 use crate::delegations::queries::query_all_network_delegations_paged;
 use crate::delegations::queries::query_delegator_delegations_paged;
 use crate::delegations::queries::query_mixnode_delegation;
-use crate::delegations::queries::query_mixnode_delegations_paged;
+use crate::delegations::queries::{
+    query_mixnode_delegations_paged, query_pending_delegation_events,
+};
 use crate::error::ContractError;
 use crate::gateways::queries::query_gateways_paged;
 use crate::gateways::queries::query_owns_gateway;
@@ -14,7 +16,7 @@ use crate::interval::queries::{
     query_current_rewarded_set_height, query_rewarded_set,
     query_rewarded_set_refresh_minimum_blocks, query_rewarded_set_update_details,
 };
-use crate::interval::storage as interval_storage;
+use crate::interval::transactions::init_epoch;
 use crate::mixnet_contract_settings::models::ContractState;
 use crate::mixnet_contract_settings::queries::{
     query_contract_settings_params, query_contract_version,
@@ -31,7 +33,7 @@ use cosmwasm_std::{
     entry_point, to_binary, Addr, Deps, DepsMut, Env, MessageInfo, QueryResponse, Response, Uint128,
 };
 use mixnet_contract_common::{
-    ContractStateParams, Delegation, ExecuteMsg, InstantiateMsg, Interval, MigrateMsg, QueryMsg,
+    ContractStateParams, Delegation, ExecuteMsg, InstantiateMsg, MigrateMsg, QueryMsg,
 };
 use time::OffsetDateTime;
 
@@ -77,13 +79,11 @@ pub fn instantiate(
 ) -> Result<Response, ContractError> {
     let rewarding_validator_address = deps.api.addr_validate(&msg.rewarding_validator_address)?;
     let state = default_initial_state(info.sender, rewarding_validator_address);
-    let rewarding_interval = Interval::init_epoch(env.clone());
+    init_epoch(deps.storage, env)?;
 
     mixnet_params_storage::CONTRACT_STATE.save(deps.storage, &state)?;
     mixnet_params_storage::LAYERS.save(deps.storage, &Default::default())?;
     rewards_storage::REWARD_POOL.save(deps.storage, &Uint128::new(INITIAL_REWARD_POOL))?;
-    interval_storage::save_epoch(deps.storage, &rewarding_interval)?;
-    interval_storage::CURRENT_REWARDED_SET_HEIGHT.save(deps.storage, &env.block.height)?;
 
     Ok(Response::default())
 }
@@ -146,18 +146,13 @@ pub fn execute(
                 deps, info, params,
             )
         }
+        #[allow(unused_variables)]
+        // FIXME: remove interval_id
         ExecuteMsg::RewardMixnode {
             identity,
             params,
             interval_id,
-        } => crate::rewards::transactions::try_reward_mixnode(
-            deps,
-            env,
-            info,
-            identity,
-            params,
-            interval_id,
-        ),
+        } => crate::rewards::transactions::try_reward_mixnode(deps, env, info, identity, params),
         ExecuteMsg::DelegateToMixnode { mix_identity } => {
             crate::delegations::transactions::try_delegate_to_mixnode(deps, env, info, mix_identity)
         }
@@ -238,9 +233,11 @@ pub fn execute(
             rewarded_set,
             expected_active_set_size,
         ),
-        ExecuteMsg::AdvanceCurrentEpoch {} => {
-            crate::interval::transactions::try_advance_epoch(env, deps.storage)
-        }
+        ExecuteMsg::AdvanceCurrentEpoch {} => crate::interval::transactions::try_advance_epoch(
+            env,
+            deps.storage,
+            info.sender.to_string(),
+        ),
         ExecuteMsg::CompoundDelegatorReward { mix_identity } => {
             crate::rewards::transactions::try_compound_delegator_reward(
                 deps,
@@ -370,13 +367,18 @@ pub fn query(deps: Deps<'_>, env: Env, msg: QueryMsg) -> Result<QueryResponse, C
             address,
             mix_identity,
         )?),
+        QueryMsg::GetPendingDelegationEvents { owner_address } => to_binary(
+            &query_pending_delegation_events(deps.storage, owner_address)?,
+        ),
     };
 
     Ok(query_res?)
 }
 
-#[entry_point]
-pub fn migrate(deps: DepsMut<'_>, _env: Env, _msg: MigrateMsg) -> Result<Response, ContractError> {
+// MIGRATE OLD DELEGATION STORAGE
+// applied on QAnet
+#[allow(dead_code)]
+fn migrate_delegations(deps: DepsMut<'_>) -> Result<(), ContractError> {
     use crate::delegations::storage::{
         DelegationIndex, DELEGATION_MIXNODE_IDX_NAMESPACE, DELEGATION_OWNER_IDX_NAMESPACE,
         DELEGATION_PK_NAMESPACE,
@@ -420,6 +422,13 @@ pub fn migrate(deps: DepsMut<'_>, _env: Env, _msg: MigrateMsg) -> Result<Respons
             &delegation,
         )?;
     }
+    Ok(())
+}
+
+#[entry_point]
+pub fn migrate(_deps: DepsMut<'_>, _env: Env, _msg: MigrateMsg) -> Result<Response, ContractError> {
+    // TODO: Uncomment for sandbox and mainnet
+    // migrate_delegations(deps)?;
 
     Ok(Default::default())
 }

contracts/mixnet/src/delegations/queries.rs

@@ -7,11 +7,24 @@ use cosmwasm_std::Order;
 use cosmwasm_std::StdResult;
 use cosmwasm_std::{Api, Deps, Storage};
 use cw_storage_plus::{Bound, PrimaryKey};
+use mixnet_contract_common::mixnode::DelegationEvent;
 use mixnet_contract_common::{
     Delegation, IdentityKey, PagedAllDelegationsResponse, PagedDelegatorDelegationsResponse,
     PagedMixDelegationsResponse,
 };
 
+pub(crate) fn query_pending_delegation_events(
+    storage: &dyn Storage,
+    owner_address: String,
+) -> Result<Vec<DelegationEvent>, ContractError> {
+    Ok(storage::PENDING_DELEGATION_EVENTS
+        .sub_prefix(owner_address.as_bytes().to_vec())
+        .range(storage, None, None, Order::Ascending)
+        .filter_map(|r| r.ok())
+        .map(|(_key, delegation_event)| delegation_event)
+        .collect::<Vec<DelegationEvent>>())
+}
+
 pub(crate) fn query_all_network_delegations_paged(
     deps: Deps<'_>,
     start_after: Option<(IdentityKey, Vec<u8>, u64)>,

contracts/mixnet/src/delegations/storage.rs

@@ -10,9 +10,9 @@ pub const DELEGATION_OWNER_IDX_NAMESPACE: &str = "dlo";
 pub const DELEGATION_MIXNODE_IDX_NAMESPACE: &str = "dlm";
 
 pub const PENDING_DELEGATION_EVENTS: Map<
-    (BlockHeight, IdentityKey, OwnerAddress),
+    (OwnerAddress, BlockHeight, IdentityKey),
     DelegationEvent,
-> = Map::new("pend");
+> = Map::new("pend2");
 
 // paged retrieval limits for all queries and transactions
 pub(crate) const DELEGATION_PAGE_MAX_LIMIT: u32 = 500;

contracts/mixnet/src/delegations/transactions.rs

@@ -40,7 +40,7 @@ pub(crate) fn _try_reconcile_all_delegation_events(
     let pending_delegation_events = PENDING_DELEGATION_EVENTS
         .range(storage, None, None, Order::Ascending)
         .filter_map(|r| r.ok())
-        .collect::<Vec<((u64, String, Vec<u8>), DelegationEvent)>>();
+        .collect::<Vec<((Vec<u8>, u64, String), DelegationEvent)>>();
 
     let mut response = Response::new();
 
@@ -193,7 +193,7 @@ pub(crate) fn _try_delegate_to_mixnode(
     }
 
     let maybe_proxy_storage = generate_storage_key(&delegate, proxy.as_ref());
-    let storage_key = (block_height, mix_identity.to_string(), maybe_proxy_storage);
+    let storage_key = (maybe_proxy_storage, block_height, mix_identity.to_string());
 
     storage::PENDING_DELEGATION_EVENTS.save(
         storage,
@@ -389,9 +389,9 @@ pub(crate) fn _try_remove_delegation_from_mixnode(
     PENDING_DELEGATION_EVENTS.save(
         deps.storage,
         (
+            delegate.as_bytes().to_vec(),
             env.block.height,
             mix_identity.to_string(),
-            delegate.as_bytes().to_vec(),
         ),
         &DelegationEvent::Undelegate(PendingUndelegate::new(
             mix_identity.to_string(),
@@ -1069,7 +1069,7 @@ mod tests {
 
             _try_reconcile_all_delegation_events(&mut deps.storage, &deps.api).unwrap();
 
-            let delegation = query_mixnode_delegation(
+            let _delegation = query_mixnode_delegation(
                 &deps.storage,
                 &deps.api,
                 identity.clone(),

contracts/mixnet/src/interval/queries.rs

@@ -10,7 +10,7 @@ use mixnet_contract_common::{
     RewardedSetUpdateDetails,
 };
 
-pub fn query_current_epoch(storage: &dyn Storage) -> Result<Option<Interval>, ContractError> {
+pub fn query_current_epoch(storage: &dyn Storage) -> Result<Interval, ContractError> {
     storage::current_epoch(storage)
 }
 

contracts/mixnet/src/interval/storage.rs

@@ -55,8 +55,10 @@ pub fn save_epoch_reward_params(
     Ok(())
 }
 
-pub fn current_epoch(storage: &dyn Storage) -> Result<Option<Interval>, ContractError> {
-    Ok(CURRENT_EPOCH.may_load(storage)?)
+pub fn current_epoch(storage: &dyn Storage) -> Result<Interval, ContractError> {
+    CURRENT_EPOCH
+        .load(storage)
+        .map_err(|_| ContractError::EpochNotInitialized)
 }
 
 pub(crate) fn save_rewarded_set(

contracts/mixnet/src/interval/transactions.rs

@@ -5,6 +5,7 @@ use super::storage;
 use crate::error::ContractError;
 use crate::error::ContractError::EpochInProgress;
 use crate::mixnet_contract_settings::storage as mixnet_params_storage;
+use crate::support::helpers::is_authorized;
 use cosmwasm_std::{DepsMut, Env, MessageInfo, Response, Storage};
 use mixnet_contract_common::events::{new_advance_interval_event, new_change_rewarded_set_event};
 use mixnet_contract_common::{IdentityKey, Interval};
@@ -64,23 +65,26 @@ pub fn try_write_rewarded_set(
     )))
 }
 
-fn init_epoch(storage: &mut dyn Storage, env: Env) -> Result<Interval, ContractError> {
+pub fn init_epoch(storage: &mut dyn Storage, env: Env) -> Result<Interval, ContractError> {
     let epoch = Interval::init_epoch(env);
     storage::save_epoch(storage, &epoch)?;
     Ok(epoch)
 }
 
-pub fn try_advance_epoch(env: Env, storage: &mut dyn Storage) -> Result<Response, ContractError> {
+pub fn try_advance_epoch(
+    env: Env,
+    storage: &mut dyn Storage,
+    sender: String,
+) -> Result<Response, ContractError> {
     // in theory, we could have just changed the state and relied on its reversal upon failed
     // execution, but better safe than sorry and do not modify the state at all unless we know
     // all checks have succeeded.
-    let current_epoch = if let Some(epoch) = storage::current_epoch(storage)? {
-        epoch
-    } else {
-        let epoch = init_epoch(storage, env)?;
-        return Ok(Response::new().add_event(new_advance_interval_event(epoch)));
-    };
 
+    // Only rewarding validator can attempt to advance epoch
+
+    is_authorized(sender, storage)?;
+
+    let current_epoch = storage::current_epoch(storage)?;
     if current_epoch.is_over(env.clone()) {
         let next_epoch = current_epoch.next_on_chain(env);
 
@@ -102,9 +106,8 @@ mod tests {
     use crate::support::tests::test_helpers;
     use cosmwasm_std::testing::{mock_env, mock_info};
     use cosmwasm_std::Timestamp;
-    use mixnet_contract_common::{Interval, RewardedSetNodeStatus};
-    use std::time::Duration;
-    use time::OffsetDateTime;
+    use mixnet_contract_common::RewardedSetNodeStatus;
+    use mixnet_params_storage::rewarding_validator_address;
 
     #[test]
     fn writing_rewarded_set() {
@@ -232,22 +235,15 @@ mod tests {
     fn advancing_epoch() {
         let mut env = mock_env();
         let mut deps = test_helpers::init_contract();
-
-        // 1609459200 = 2021-01-01
-        // 1640995200 = 2022-01-01
-        // 1641081600 = 2022-01-02
-        // 1643673600 = 2022-02-01
-        // 1672531200 = 2023-01-01
+        let sender = rewarding_validator_address(&deps.storage).unwrap();
 
         let _current_epoch = init_epoch(&mut deps.storage, env.clone()).unwrap();
 
         // Works as its after the current epoch
         env.block.time = Timestamp::from_seconds(1641081600);
-        assert!(try_advance_epoch(env.clone(), deps.as_mut().storage).is_ok());
+        assert!(try_advance_epoch(env.clone(), deps.as_mut().storage, sender.clone()).is_ok());
 
-        let current_epoch = crate::interval::storage::current_epoch(&mut deps.storage)
-            .unwrap()
-            .unwrap();
+        let current_epoch = crate::interval::storage::current_epoch(&mut deps.storage).unwrap();
 
         // same if the current blocktime is set to BEFORE the first interval has even begun
         // (say we decided to set the first interval to be some time in the future at initialisation)
@@ -258,7 +254,7 @@ mod tests {
                 epoch_start: current_epoch.start_unix_timestamp(),
                 epoch_end: current_epoch.end_unix_timestamp()
             }),
-            try_advance_epoch(env.clone(), deps.as_mut().storage)
+            try_advance_epoch(env.clone(), deps.as_mut().storage, sender.clone(),)
         );
 
         // works otherwise
@@ -271,7 +267,7 @@ mod tests {
             Response::new().add_event(new_advance_interval_event(expected_new_epoch));
         assert_eq!(
             Ok(expected_response),
-            try_advance_epoch(env.clone(), deps.as_mut().storage)
+            try_advance_epoch(env.clone(), deps.as_mut().storage, sender)
         );
 
         // interval way back in the past (i.e. 'somebody' failed to advance it for a long time)
@@ -280,9 +276,10 @@ mod tests {
         let expected_new_epoch = current_epoch.next_on_chain(env.clone());
         let expected_response =
             Response::new().add_event(new_advance_interval_event(expected_new_epoch));
+        let sender = rewarding_validator_address(&deps.storage).unwrap();
         assert_eq!(
             Ok(expected_response),
-            try_advance_epoch(env.clone(), deps.as_mut().storage)
+            try_advance_epoch(env.clone(), deps.as_mut().storage, sender)
         );
     }
 }