Partition proactive refresh

crypto/server/src/signing_client/api.rs

@@ -16,7 +16,9 @@ use entropy_protocol::{
 };
 use parity_scale_codec::Encode;
 
-use entropy_shared::{KeyVisibility, OcwMessageProactiveRefresh, SETUP_TIMEOUT_SECONDS};
+use entropy_shared::{
+    KeyVisibility, OcwMessageProactiveRefresh, REFRESHES_PER_SESSION, SETUP_TIMEOUT_SECONDS,
+};
 use kvdb::kv_manager::{
     helpers::{deserialize, serialize as key_serialize},
     KvManager,
@@ -67,6 +69,7 @@ pub async fn proactive_refresh(
     // TODO batch the network keys into smaller groups per session
     let all_keys =
         get_all_keys(&api, &rpc).await.map_err(|e| ProtocolErr::ValidatorErr(e.to_string()))?;
+    let proactive_refresh_keys = partition_all_keys(ocw_data.refreshes_done, all_keys);
     let (subgroup, stash_address) = get_subgroup(&api, &rpc, &signer)
         .await
         .map_err(|e| ProtocolErr::UserError(e.to_string()))?;
@@ -77,7 +80,7 @@ pub async fn proactive_refresh(
     let subxt_signer = get_subxt_signer(&app_state.kv_store)
         .await
         .map_err(|e| ProtocolErr::UserError(e.to_string()))?;
-    for key in all_keys {
+    for key in proactive_refresh_keys {
         let sig_request_address = AccountId32::from_str(&key).map_err(ProtocolErr::StringError)?;
         let key_visibility =
             get_key_visibility(&api, &rpc, &sig_request_address.clone().into()).await.unwrap();
@@ -236,7 +239,7 @@ pub async fn validate_proactive_refresh(
         })?;
 
     let mut hasher_chain_data = Blake2s256::new();
-    hasher_chain_data.update(ocw_data.validators_info.encode());
+    hasher_chain_data.update(ocw_data.encode());
     let chain_data_hash = hasher_chain_data.finalize();
     let mut hasher_verifying_data = Blake2s256::new();
     hasher_verifying_data.update(proactive_info.encode());
@@ -252,3 +255,50 @@ pub async fn validate_proactive_refresh(
     kv_manager.kv().put(reservation, latest_block_number.to_be_bytes().to_vec()).await?;
     Ok(())
 }
+
+/// Partitions all registered keys into a subset of the network (REFRESHES_PRE_SESSION)
+/// Currently rotates between a moving batch of all keys.
+///
+/// See https://github.com/entropyxyz/entropy-core/issues/510 for some issues which exist
+/// around the scaling of this function.
+pub fn partition_all_keys(refreshes_done: u32, all_keys: Vec<String>) -> Vec<String> {
+    let all_keys_length = all_keys.len() as u32;
+
+    // just return all keys no need to partition network
+    if REFRESHES_PER_SESSION > all_keys_length {
+        return all_keys;
+    }
+
+    let mut refresh_keys: Vec<String> = vec![];
+
+    // handles early on refreshes before refreshes done > all keys
+    if refreshes_done + REFRESHES_PER_SESSION <= all_keys_length {
+        let lower = refreshes_done as usize;
+        let upper = (refreshes_done + REFRESHES_PER_SESSION) as usize;
+        refresh_keys = all_keys[lower..upper].to_vec();
+    }
+
+    // normalize refreshes done down to a partition of the network
+    let normalized_refreshes_done = refreshes_done % all_keys_length;
+
+    if normalized_refreshes_done + REFRESHES_PER_SESSION <= all_keys_length {
+        let lower = normalized_refreshes_done as usize;
+        let upper = (normalized_refreshes_done + REFRESHES_PER_SESSION) as usize;
+        refresh_keys = all_keys[lower..upper].to_vec();
+    }
+
+    // handles if number does not perfectly fit
+    // loops around the partiton adding the beginning of the network to the end
+    if normalized_refreshes_done + REFRESHES_PER_SESSION > all_keys_length {
+        let lower = normalized_refreshes_done as usize;
+        let upper = all_keys.len();
+        refresh_keys = all_keys[lower..upper].to_vec();
+
+        let leftover =
+            (REFRESHES_PER_SESSION - (all_keys_length - normalized_refreshes_done)) as usize;
+        let mut post_turnaround_keys = all_keys[0..leftover].to_vec();
+        refresh_keys.append(&mut post_turnaround_keys);
+    }
+
+    refresh_keys
+}

crypto/server/src/signing_client/tests.rs

@@ -1,4 +1,4 @@
-use super::api::validate_proactive_refresh;
+use super::api::{partition_all_keys, validate_proactive_refresh};
 use crate::{
     chain_api::{get_api, get_rpc},
     helpers::{
@@ -79,7 +79,7 @@ async fn test_proactive_refresh() {
         },
     ];
 
-    let mut ocw_message = OcwMessageProactiveRefresh { validators_info };
+    let mut ocw_message = OcwMessageProactiveRefresh { validators_info, refreshes_done: 0 };
 
     let test_fail_incorrect_data =
         submit_transaction_requests(validator_ips.clone(), ocw_message.clone()).await;
@@ -182,7 +182,7 @@ async fn test_proactive_refresh_validation_fail() {
     ];
 
     let block_number = rpc.chain_get_header(None).await.unwrap().unwrap().number + 1;
-    let ocw_message = OcwMessageProactiveRefresh { validators_info };
+    let ocw_message = OcwMessageProactiveRefresh { validators_info, refreshes_done: 0 };
     run_to_block(&rpc, block_number).await;
 
     // manipulates kvdb to get to repeated data error
@@ -196,3 +196,27 @@ async fn test_proactive_refresh_validation_fail() {
     assert_eq!(err_stale_data, Err("Data is repeated".to_string()));
     clean_tests();
 }
+
+#[test]
+fn test_partition_all_keys() {
+    let all_keys: Vec<String> = (1..=25).map(|num| num.to_string()).collect();
+
+    let result_normal_10 = partition_all_keys(2, all_keys.clone());
+    assert_eq!(result_normal_10, all_keys[2..12].to_vec());
+
+    let result_next_set = partition_all_keys(12, all_keys.clone());
+    assert_eq!(result_next_set, all_keys[12..22].to_vec());
+
+    let wrap_around_partial = partition_all_keys(23, all_keys.clone());
+    let mut wrap_around_partial_vec = all_keys[23..25].to_vec();
+    wrap_around_partial_vec.append(&mut all_keys[0..8].to_vec());
+    assert_eq!(wrap_around_partial, wrap_around_partial_vec);
+
+    let result_larger = partition_all_keys(32, all_keys.clone());
+    assert_eq!(result_larger, all_keys[7..17].to_vec());
+
+    let wrap_around_partial_larger = partition_all_keys(42, all_keys.clone());
+    let mut wrap_around_partial_larger_vec = all_keys[17..25].to_vec();
+    wrap_around_partial_larger_vec.append(&mut all_keys[0..2].to_vec());
+    assert_eq!(wrap_around_partial_larger, wrap_around_partial_larger_vec);
+}

crypto/shared/src/lib.rs

@@ -18,3 +18,6 @@ pub const PRUNE_BLOCK: u32 = 14400;
 
 /// Timeout for validators to wait for other validators to join protocol committees
 pub const SETUP_TIMEOUT_SECONDS: u64 = 20;
+
+/// The amount of proactive refreshes we do per session
+pub const REFRESHES_PER_SESSION: u32 = 10;

crypto/shared/src/types.rs

@@ -72,4 +72,5 @@ pub struct OcwMessageDkg {
 )]
 pub struct OcwMessageProactiveRefresh {
     pub validators_info: Vec<ValidatorInfo>,
+    pub refreshes_done: u32,
 }

pallets/propagation/src/lib.rs

@@ -126,8 +126,7 @@ pub mod pallet {
 
         pub fn post_proactive_refresh(_block_number: BlockNumberFor<T>) -> Result<(), http::Error> {
             let refresh_info = pallet_staking_extension::Pallet::<T>::proactive_refresh();
-
-            if refresh_info.is_empty() {
+            if refresh_info.validators_info.is_empty() {
                 return Ok(());
             }
 
@@ -138,20 +137,12 @@ pub mod pallet {
             let url = str::from_utf8(&from_local)
                 .unwrap_or("http://localhost:3001/signer/proactive_refresh");
 
-            let (servers_info, _i) =
-                pallet_relayer::Pallet::<T>::get_validator_info().unwrap_or_default();
-            let validators_info = servers_info
-                .iter()
-                .map(|server_info| ValidatorInfo {
-                    x25519_public_key: server_info.x25519_public_key,
-                    ip_address: server_info.endpoint.clone(),
-                    tss_account: server_info.tss_account.encode(),
-                })
-                .collect::<Vec<_>>();
-
-            log::warn!("propagation::post proactive refresh: {:?}", &[validators_info.encode()]);
+            let req_body = OcwMessageProactiveRefresh {
+                validators_info: refresh_info.validators_info,
+                refreshes_done: refresh_info.refreshes_done,
+            };
+            log::warn!("propagation::post proactive refresh: {:?}", &[req_body.encode()]);
 
-            let req_body = OcwMessageProactiveRefresh { validators_info };
             // We construct the request
             // important: the header->Content-Type must be added and match that of the receiving
             // party!!

pallets/propagation/src/tests.rs

@@ -3,6 +3,7 @@ use std::sync::Arc;
 use codec::Encode;
 use entropy_shared::{KeyVisibility, ValidatorInfo};
 use frame_support::{assert_ok, traits::OnInitialize};
+use pallet_staking_extension::RefreshInfo;
 use sp_core::offchain::{testing, OffchainDbExt, OffchainWorkerExt, TransactionPoolExt};
 use sp_io::TestExternalities;
 use sp_keystore::{testing::MemoryKeystore, KeystoreExt};
@@ -47,10 +48,8 @@ fn knows_how_to_mock_several_http_calls() {
             sent: true,
             response: Some([].to_vec()),
             body: [
-                8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
-                0, 0, 0, 0, 0, 4, 20, 32, 7, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
-                0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 4, 40, 32, 8, 0, 0,
-                0, 0, 0, 0, 0,
+                4, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+                0, 0, 0, 0, 0, 20, 16, 116, 101, 115, 116, 20, 16, 116, 101, 115, 116, 0, 0, 0, 0,
             ]
             .to_vec(),
             ..Default::default()
@@ -71,14 +70,18 @@ fn knows_how_to_mock_several_http_calls() {
         assert_eq!(Relayer::dkg(3).len(), 0);
 
         Propagation::post_proactive_refresh(6).unwrap();
-        pallet_staking_extension::ProactiveRefresh::<Test>::put(vec![ValidatorInfo {
-            x25519_public_key: [0u8; 32],
-            ip_address: "test".encode(),
-            tss_account: "test".encode(),
-        }]);
+        let ocw_message = RefreshInfo {
+            validators_info: vec![ValidatorInfo {
+                x25519_public_key: [0u8; 32],
+                ip_address: "test".encode(),
+                tss_account: "test".encode(),
+            }],
+            refreshes_done: 0,
+        };
+        pallet_staking_extension::ProactiveRefresh::<Test>::put(ocw_message);
         Propagation::post_proactive_refresh(6).unwrap();
         Propagation::on_initialize(6);
-        assert_eq!(Staking::proactive_refresh(), vec![]);
+        assert_eq!(Staking::proactive_refresh(), RefreshInfo::default());
     })
 }
 

pallets/staking/src/lib.rs

@@ -86,6 +86,12 @@ pub mod pallet {
         pub x25519_public_key: X25519PublicKey,
         pub endpoint: TssServerURL,
     }
+    /// Info that is requiered to do a proactive refresh
+    #[derive(Clone, Encode, Decode, Eq, PartialEq, RuntimeDebug, TypeInfo, Default)]
+    pub struct RefreshInfo {
+        pub validators_info: Vec<ValidatorInfo>,
+        pub refreshes_done: u32,
+    }
 
     #[pallet::pallet]
     #[pallet::without_storage_info]
@@ -140,9 +146,10 @@ pub mod pallet {
         ValueQuery,
     >;
 
+    /// A trigger for the proactive refresh OCW
     #[pallet::storage]
     #[pallet::getter(fn proactive_refresh)]
-    pub type ProactiveRefresh<T: Config> = StorageValue<_, Vec<ValidatorInfo>, ValueQuery>;
+    pub type ProactiveRefresh<T: Config> = StorageValue<_, RefreshInfo, ValueQuery>;
 
     #[pallet::genesis_config]
     #[derive(DefaultNoBound)]
@@ -175,8 +182,11 @@ pub mod pallet {
                     IsValidatorSynced::<T>::insert(validator_id, true);
                 }
             }
-
-            ProactiveRefresh::<T>::put(self.proactive_refresh_validators.clone());
+            let refresh_info = RefreshInfo {
+                validators_info: self.proactive_refresh_validators.clone(),
+                refreshes_done: 0,
+            };
+            ProactiveRefresh::<T>::put(refresh_info);
         }
     }
     // Errors inform users that something went wrong.
@@ -350,6 +360,7 @@ pub mod pallet {
         pub fn new_session_handler(
             validators: &[<T as pallet_session::Config>::ValidatorId],
         ) -> Result<(), DispatchError> {
+            // TODO add back in refresh trigger and refreshed counter https://github.com/entropyxyz/entropy-core/issues/511
             // Init a 2D Vec where indices and values represent subgroups and validators,
             // respectively.
             let mut new_validators_set: Vec<Vec<<T as pallet_session::Config>::ValidatorId>> =