update offchain worker logic

Signed-off-by: Uncle Stinky <uncle.stinky@ghostchain.io>

pallets/slow-clap/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ghost-slow-clap"
-version = "0.3.53"
+version = "0.3.54"
 description = "Applause protocol for the EVM bridge"
 license.workspace = true
 authors.workspace = true

pallets/slow-clap/src/lib.rs

@@ -23,7 +23,7 @@ use sp_core::H256;
 use sp_runtime::{
     offchain::{
         self as rt_offchain,
-        storage::{MutateStorageError, StorageRetrievalError, StorageValueRef},
+        storage::StorageValueRef,
         storage_lock::{StorageLock, Time},
         HttpError,
     },
@@ -97,8 +97,6 @@ pub struct SessionAuthorityInfo {
 
 #[cfg_attr(test, derive(PartialEq))]
 enum OffchainErr<NetworkId> {
-    FailedSigning,
-    SubmitTransaction,
     HttpJsonParsingError,
     HttpBytesParsingError,
     HttpRequestError(HttpError),
@@ -106,38 +104,65 @@ enum OffchainErr<NetworkId> {
     HttpResponseNotOk(u16),
     ErrorInEvmResponse,
     NoStoredNetworks,
-    NotValidator,
+    NoEndpointAvailable(NetworkId),
     StorageRetrievalError(NetworkId),
-    ConcurrentModificationError(NetworkId),
     UtxoNotImplemented(NetworkId),
     UnknownNetworkType(NetworkId),
     OffchainTimeoutPeriod(NetworkId),
-    TooManyRequests(NetworkId),
 }
 
 impl<NetworkId: core::fmt::Debug> core::fmt::Debug for OffchainErr<NetworkId> {
     fn fmt(&self, fmt: &mut core::fmt::Formatter) -> core::fmt::Result {
         match *self {
-            OffchainErr::FailedSigning => write!(fmt, "Failed to sign clap."),
+            OffchainErr::HttpJsonParsingError => {
-            OffchainErr::SubmitTransaction => write!(fmt, "Failed to submit transaction."),
+                write!(fmt, "Failed to parse evm response as JSON.")
-            OffchainErr::HttpJsonParsingError => write!(fmt, "Failed to parse evm response as JSON."),
+            }
-            OffchainErr::HttpBytesParsingError => write!(fmt, "Failed to parse evm response as bytes."),
+            OffchainErr::HttpBytesParsingError => {
+                write!(fmt, "Failed to parse evm response as bytes.")
+            }
             OffchainErr::HttpRequestError(http_error) => match http_error {
-                HttpError::DeadlineReached => write!(fmt, "Requested action couldn't been completed within a deadline."),
+                HttpError::DeadlineReached => write!(
-                HttpError::IoError => write!(fmt, "There was an IO error while processing the request."),
+                    fmt,
-                HttpError::Invalid => write!(fmt, "The ID of the request is invalid in this context."),
+                    "Requested action couldn't been completed within a deadline."
+                ),
+                HttpError::IoError => {
+                    write!(fmt, "There was an IO error while processing the request.")
+                }
+                HttpError::Invalid => {
+                    write!(fmt, "The ID of the request is invalid in this context.")
+                }
             },
-            OffchainErr::ConcurrentModificationError(ref network_id) => write!(fmt, "The underlying DB failed to update due to a concurrent modification for network #{:?}.", network_id),
+            OffchainErr::StorageRetrievalError(ref network_id) => write!(
-            OffchainErr::StorageRetrievalError(ref network_id) => write!(fmt, "Storage value found for network #{:?} but it's undecodable.", network_id),
+                fmt,
+                "Storage value found for network #{:?} but it's undecodable.",
+                network_id
+            ),
             OffchainErr::RequestUncompleted => write!(fmt, "Failed to complete request."),
-            OffchainErr::HttpResponseNotOk(code) => write!(fmt, "Http response returned code {:?}.", code),
+            OffchainErr::HttpResponseNotOk(code) => {
+                write!(fmt, "Http response returned code {:?}.", code)
+            }
             OffchainErr::ErrorInEvmResponse => write!(fmt, "Error in evm reponse."),
-            OffchainErr::NoStoredNetworks => write!(fmt, "No networks stored for the offchain slow claps."),
+            OffchainErr::NoStoredNetworks => {
-            OffchainErr::NotValidator => write!(fmt, "Not a validator for slow clap, `--validator` flag needed."),
+                write!(fmt, "No networks stored for the offchain slow claps.")
-            OffchainErr::UtxoNotImplemented(ref network_id) => write!(fmt, "Network #{:?} is marked as UTXO, which is not implemented yet.", network_id),
+            }
-            OffchainErr::UnknownNetworkType(ref network_id) => write!(fmt, "Unknown type for network #{:?}.", network_id),
+            OffchainErr::NoEndpointAvailable(ref network_id) => write!(
-            OffchainErr::OffchainTimeoutPeriod(ref network_id) => write!(fmt, "Offchain request should be in-flight for network #{:?}.", network_id),
+                fmt,
-            OffchainErr::TooManyRequests(ref network_id) => write!(fmt, "Too many requests over RPC endpoint for network #{:?}.", network_id),
+                "No RPC endpoint available for network #{:?}.",
+                network_id
+            ),
+            OffchainErr::UtxoNotImplemented(ref network_id) => write!(
+                fmt,
+                "Network #{:?} is marked as UTXO, which is not implemented yet.",
+                network_id
+            ),
+            OffchainErr::UnknownNetworkType(ref network_id) => {
+                write!(fmt, "Unknown type for network #{:?}.", network_id)
+            }
+            OffchainErr::OffchainTimeoutPeriod(ref network_id) => write!(
+                fmt,
+                "Offchain request should be in-flight for network #{:?}.",
+                network_id
+            ),
         }
     }
 }
@@ -359,10 +384,7 @@ pub mod pallet {
     #[pallet::hooks]
     impl<T: Config> Hooks<BlockNumberFor<T>> for Pallet<T> {
         fn offchain_worker(now: BlockNumberFor<T>) {
-            match Self::start_slow_clapping(now) {
+            if let Err(e) = Self::start_slow_clapping(now) {
-                Ok(iter) => {
-                    for result in iter.into_iter() {
-                        if let Err(e) = result {
                 log::info!(
                     target: LOG_TARGET,
                     "👏 Skipping slow clap at {:?}: {:?}",
@@ -372,15 +394,6 @@ pub mod pallet {
             }
         }
     }
-                Err(e) => log::info!(
-                    target: LOG_TARGET,
-                    "👏 Could not start slow clap at {:?}: {:?}",
-                    now,
-                    e,
-                ),
-            }
-        }
-    }
 
     #[pallet::validate_unsigned]
     impl<T: Config> ValidateUnsigned for Pallet<T> {
@@ -680,13 +693,7 @@ impl<T: Config> Pallet<T> {
         Ok(())
     }
 
-    fn start_slow_clapping(
+    fn start_slow_clapping(block_number: BlockNumberFor<T>) -> OffchainResult<T, ()> {
-        block_number: BlockNumberFor<T>,
-    ) -> OffchainResult<T, impl Iterator<Item = OffchainResult<T, ()>>> {
-        sp_io::offchain::is_validator()
-            .then(|| ())
-            .ok_or(OffchainErr::NotValidator)?;
-
         let session_index = T::ValidatorSet::session_index();
         let networks_len = T::NetworkDataHandler::iter().count();
         let network_in_use = T::NetworkDataHandler::iter()
@@ -701,7 +708,6 @@ impl<T: Config> Pallet<T> {
 
         let network_id_encoded = network_in_use.0.encode();
 
-        let last_timestamp_key = Self::create_storage_key(b"last-timestamp-", &network_id_encoded);
         let rate_limit_delay_key = Self::create_storage_key(b"rate-limit-", &network_id_encoded);
         let rate_limit_delay = Self::read_persistent_offchain_storage(
             &rate_limit_delay_key,
@@ -710,51 +716,24 @@ impl<T: Config> Pallet<T> {
 
         let network_lock_key = Self::create_storage_key(b"network-lock-", &network_id_encoded);
         let block_until =
-            rt_offchain::Duration::from_millis(networks_len as u64 * FETCH_TIMEOUT_PERIOD);
+            rt_offchain::Duration::from_millis(rate_limit_delay.max(FETCH_TIMEOUT_PERIOD));
         let mut network_lock = StorageLock::<Time>::with_deadline(&network_lock_key, block_until);
 
-        network_lock
+        let _lock_guard = network_lock
             .try_lock()
             .map_err(|_| OffchainErr::OffchainTimeoutPeriod(network_in_use.0))?;
 
-        StorageValueRef::persistent(&last_timestamp_key)
+        log::info!(
-            .mutate(
+            target: LOG_TARGET,
-                |result_timestamp: Result<Option<u64>, StorageRetrievalError>| {
+            "🧐 Offchain worker started for network #{:?} at block #{:?}",
-                    let current_timestmap = sp_io::offchain::timestamp().unix_millis();
-                    match result_timestamp {
-                        Ok(option_timestamp) => match option_timestamp {
-                            Some(stored_timestamp) if stored_timestamp > current_timestmap => {
-                                Err(OffchainErr::TooManyRequests(network_in_use.0).into())
-                            }
-                            _ => Ok(current_timestmap.saturating_add(rate_limit_delay)),
-                        },
-                        Err(_) => Err(OffchainErr::StorageRetrievalError(network_in_use.0).into()),
-                    }
-                },
-            )
-            .map_err(|error| match error {
-                MutateStorageError::ValueFunctionFailed(offchain_error) => offchain_error,
-                MutateStorageError::ConcurrentModification(_) => {
-                    OffchainErr::ConcurrentModificationError(network_in_use.0).into()
-                }
-            })?;
-
-        Ok(
-            Self::local_authorities(&session_index).map(move |(authority_index, authority_key)| {
-                Self::do_evm_claps_or_save_block(
-                    authority_index,
-                    authority_key,
-                    session_index,
             network_in_use.0,
-                    &network_in_use.1,
+            block_number,
-                )
+        );
-            }),
+
-        )
+        Self::do_evm_claps_or_save_block(session_index, network_in_use.0, &network_in_use.1)
     }
 
     fn do_evm_claps_or_save_block(
-        authority_index: AuthIndex,
-        authority_key: T::AuthorityId,
         session_index: SessionIndex,
         network_id: NetworkIdOf<T>,
         network_data: &NetworkData,
@@ -778,36 +757,26 @@ impl<T: Config> Pallet<T> {
         let random_number = <u32>::decode(&mut TrailingZeroInput::new(random_seed.as_ref()))
             .expect("input is padded with zeroes; qed");
 
-        let rpc_endpoint = if stored_endpoints.len() > 0 {
+        let random_index = (random_number as usize)
-            stored_endpoints
-                .iter()
-                .nth(
-                    (random_number as usize)
             .checked_rem(stored_endpoints.len())
-                        .unwrap_or_default(),
+            .unwrap_or_default();
-                )
+
-                .expect("stored endpoint should be non empty; qed")
+        let endpoints = if !stored_endpoints.is_empty() {
+            &stored_endpoints
         } else {
-            network_data
+            &network_data.default_endpoints
-                .default_endpoints
-                .iter()
-                .nth(
-                    (random_number as usize)
-                        .checked_rem(network_data.default_endpoints.len())
-                        .unwrap_or_default(),
-                )
-                .expect("default endpoint should be non empty; qed")
         };
 
-        StorageValueRef::persistent(&block_number_key)
+        let rpc_endpoint = endpoints
-            .mutate(
+            .get(random_index)
-                |result_block_range: Result<Option<(u64, u64)>, StorageRetrievalError>| {
+            .ok_or(OffchainErr::NoEndpointAvailable(network_id))?;
-                    match result_block_range {
+
-                        Ok(maybe_block_range) => {
+        let maybe_block_range: Option<(u64, u64)> = StorageValueRef::persistent(&block_number_key)
+            .get()
+            .map_err(|_| OffchainErr::StorageRetrievalError(network_id))?;
+
         let request_body = match maybe_block_range {
-                                Some((from_block, to_block))
+            Some((from_block, to_block)) if from_block < to_block.saturating_sub(1) => {
-                                    if from_block < to_block.saturating_sub(1) =>
-                                {
                 Self::prepare_request_body_for_latest_transfers(
                     from_block,
                     to_block.saturating_sub(1),
@@ -817,72 +786,38 @@ impl<T: Config> Pallet<T> {
             _ => Self::prepare_request_body_for_latest_block(network_data),
         };
 
-                            let response_bytes =
+        let response_bytes = Self::fetch_from_remote(&rpc_endpoint, &request_body)?;
-                                Self::fetch_from_remote(&rpc_endpoint, &request_body)?;
 
         match network_data.network_type {
             NetworkType::Evm => {
-                                    let maybe_new_evm_block = Self::apply_evm_response(
+                let maybe_new_evm_block =
-                                        &response_bytes,
+                    Self::apply_evm_response(&response_bytes, session_index, network_id)?;
-                                        authority_index,
-                                        authority_key,
-                                        session_index,
-                                        network_id,
-                                    )?;
-
                 let estimated_block = maybe_new_evm_block
-                                        .map(|new_evm_block| {
+                    .map(|new_evm_block| new_evm_block.saturating_sub(network_data.finality_delay))
-                                            new_evm_block
-                                                .saturating_sub(network_data.finality_delay)
-                                        })
                     .unwrap_or_default();
 
-                                    Ok(match maybe_block_range {
+                let new_block_range = match maybe_block_range {
                     Some((from_block, to_block)) => match maybe_new_evm_block {
-                                            Some(_) if from_block.le(&to_block) => {
+                        Some(_) if from_block.le(&to_block) => (
-                                                let adjusted_to_block = estimated_block
+                            from_block,
-                                                    .checked_sub(from_block)
+                            Self::adjust_to_block(estimated_block, from_block, max_block_distance),
-                                                    .map(|current_distance| {
+                        ),
-                                                        current_distance
-                                                            .le(&max_block_distance)
-                                                            .then(|| estimated_block)
-                                                    })
-                                                    .flatten()
-                                                    .unwrap_or(
-                                                        from_block
-                                                            .saturating_add(max_block_distance)
-                                                            .min(estimated_block),
-                                                    );
-                                                (from_block, adjusted_to_block)
-                                            }
                         _ => (to_block, to_block),
                     },
                     None => (estimated_block, estimated_block),
-                                    })
+                };
-                                }
+
-                                NetworkType::Utxo => {
+                StorageValueRef::persistent(&block_number_key).set(&new_block_range);
-                                    Err(OffchainErr::UtxoNotImplemented(network_id).into())
+
+                Ok(())
             }
+            NetworkType::Utxo => Err(OffchainErr::UtxoNotImplemented(network_id).into()),
             _ => Err(OffchainErr::UnknownNetworkType(network_id).into()),
         }
     }
-                        Err(_) => Err(OffchainErr::StorageRetrievalError(network_id).into()),
-                    }
-                },
-            )
-            .map_err(|error| match error {
-                MutateStorageError::ValueFunctionFailed(offchain_error) => offchain_error,
-                MutateStorageError::ConcurrentModification(_) => {
-                    OffchainErr::ConcurrentModificationError(network_id).into()
-                }
-            })
-            .map(|_| ())
-    }
 
     fn apply_evm_response(
         response_bytes: &[u8],
-        authority_index: AuthIndex,
-        authority_key: T::AuthorityId,
         session_index: SessionIndex,
         network_id: NetworkIdOf<T>,
     ) -> OffchainResult<T, Option<u64>> {
@@ -897,6 +832,12 @@ impl<T: Config> Pallet<T> {
                 Ok(Some(new_evm_block))
             }
             EvmResponseType::TransactionLogs(evm_logs) => {
+                if sp_io::offchain::is_validator() {
+                    log::info!(target: LOG_TARGET, "🧐 Not a validator; no claps available");
+                    return Ok(None);
+                }
+
+                for (authority_index, authority_key) in Self::local_authorities(&session_index) {
                     if ClapsInSession::<T>::get(&session_index)
                         .get(&authority_index)
                         .map(|info| info.disabled)
@@ -908,7 +849,7 @@ impl<T: Config> Pallet<T> {
                             authority_index,
                             session_index
                         );
-                    return Ok(None);
+                        continue;
                     }
 
                     let claps: Vec<_> = evm_logs
@@ -943,14 +884,7 @@ impl<T: Config> Pallet<T> {
                         network_id,
                     );
 
-                for clap in claps {
+                    Self::sign_and_submit_claps(authority_key, authority_index, network_id, &claps);
-                    let signature = authority_key
-                        .sign(&clap.encode())
-                        .ok_or(OffchainErr::FailedSigning)?;
-                    let call = Call::slow_clap { clap, signature };
-
-                    SubmitTransaction::<T, Call<T>>::submit_unsigned_transaction(call.into())
-                        .map_err(|_| OffchainErr::SubmitTransaction)?;
                 }
 
                 Ok(None)
@@ -958,6 +892,63 @@ impl<T: Config> Pallet<T> {
         }
     }
 
+    fn sign_and_submit_claps(
+        authority_key: T::AuthorityId,
+        authority_index: AuthIndex,
+        network_id: NetworkIdOf<T>,
+        claps: &Vec<Clap<T::AccountId, NetworkIdOf<T>, BalanceOf<T>>>,
+    ) {
+        for (clap_index, clap) in claps.iter().enumerate() {
+            let signature = match authority_key.sign(&clap.encode()) {
+                Some(signature) => signature,
+                None => {
+                    log::info!(
+                        target: LOG_TARGET,
+                        "🧐 Failed to sign clap #{:?} from authority #{:?} for network {:?}",
+                        clap_index,
+                        authority_index,
+                        network_id,
+                    );
+                    continue;
+                }
+            };
+
+            let call = Call::slow_clap {
+                clap: clap.clone(),
+                signature,
+            };
+
+            if let Err(e) =
+                SubmitTransaction::<T, Call<T>>::submit_unsigned_transaction(call.into())
+            {
+                log::info!(
+                    target: LOG_TARGET,
+                    "🧐 Failed to submit clap #{:?} from authority #{:?} for network {:?}: {:?}",
+                    clap_index,
+                    authority_index,
+                    network_id,
+                    e,
+                );
+            }
+        }
+    }
+
+    fn adjust_to_block(estimated_block: u64, from_block: u64, max_block_distance: u64) -> u64 {
+        let fallback_value = from_block
+            .saturating_add(max_block_distance)
+            .min(estimated_block);
+
+        estimated_block
+            .checked_sub(from_block)
+            .map(|current_distance| {
+                current_distance
+                    .le(&max_block_distance)
+                    .then(|| estimated_block)
+            })
+            .flatten()
+            .unwrap_or(fallback_value)
+    }
+
     fn local_authorities(
         session_index: &SessionIndex,
     ) -> impl Iterator<Item = (u32, T::AuthorityId)> {

pallets/slow-clap/src/tests.rs

@@ -304,6 +304,16 @@ fn request_body_is_correct_for_get_logs() {
     });
 }
 
+#[test]
+fn should_correctly_adjust_to_block() {
+    assert_eq!(SlowClap::adjust_to_block(420, 69, 1337), 420);
+    assert_eq!(SlowClap::adjust_to_block(420, 1337, 69), 420);
+    assert_eq!(SlowClap::adjust_to_block(1337, 420, 69), 489);
+    assert_eq!(SlowClap::adjust_to_block(1337, 69, 420), 489);
+    assert_eq!(SlowClap::adjust_to_block(69, 1337, 420), 69);
+    assert_eq!(SlowClap::adjust_to_block(69, 420, 1337), 69);
+}
+
 #[test]
 fn should_make_http_call_for_block_number() {
     let (offchain, state) = TestOffchainExt::new();
@@ -361,8 +371,7 @@ fn should_make_http_call_and_parse_block_number() {
 
         let request_body = SlowClap::prepare_request_body_for_latest_block(&network_data);
         let raw_response = SlowClap::fetch_from_remote(&rpc_endpoint, &request_body)?;
-        let maybe_evm_block_number =
+        let maybe_evm_block_number = SlowClap::apply_evm_response(&raw_response, 420, 1)?;
-            SlowClap::apply_evm_response(&raw_response, 69, Default::default(), 420, 1)?;
 
         assert_eq!(maybe_evm_block_number, Some(20335745));
         Ok(())
@@ -402,13 +411,8 @@ fn should_make_http_call_and_parse_logs() {
             EvmResponseType::TransactionLogs(evm_logs) => assert_eq!(evm_logs.len(), 2),
         }
 
-        let maybe_evm_block_number = SlowClap::apply_evm_response(
+        let maybe_evm_block_number =
-            &raw_response,
+            SlowClap::apply_evm_response(&raw_response, session_index, network_id)?;
-            1,
-            UintAuthorityId::from(2),
-            session_index,
-            network_id,
-        )?;
 
         assert_eq!(maybe_evm_block_number, None);
         Ok(())