Fix formulate reorg and event workers (#374)

* Test multiple chunks in formulate reorg

* Fix broken StorageLock...

The StorageLocks all have a default deadline, they don't just remain locked.
For some reason (yet to debug), the worker calls are taking a long time / timing out, and due to the broken lock additional worker processes are piling up.

* Dont give up formulating a reorg, even if we are missing local data

* Back to StorageLock with increased deadline

Each WASM worker runs in its own thread and does not have shared memory, so the previous fix only worked for native.

Setting to 120s deadline should work most of the time, and when it doesn't it's not the end of the world to ignore a few irrelevant blocks, just inefficient if it happens too often.
The other side effect is that the storage lock persists across restarts, which is the downside of making it too long.

* Rewrite formulate reorg to process forward/reverse in lockstep

Also fix the handling for traversing before the first block, but err out if the worker is otherwise missing data.

* Bump version

chains/build_spec.js

@@ -1,4 +1,4 @@
-#!env node
+#!/usr/bin/env node
 const child_process = require('child_process');
 const fs = require('fs').promises;
 const path = require('path');

ethereum-client/src/lib.rs

@@ -6,7 +6,7 @@ use hex_buffer_serde::{ConstHex, ConstHexForm};
 use sp_runtime::offchain::{http, Duration};
 use sp_runtime_interface::pass_by::PassByCodec;
 
-use our_std::{debug, error, info, warn, Deserialize, RuntimeDebug, Serialize};
+use our_std::{error, info, trace, warn, Deserialize, RuntimeDebug, Serialize};
 use types_derive::{type_alias, Types};
 
 pub mod events;
@@ -182,7 +182,7 @@ pub fn send_rpc(
         "id":1
     })
     .to_string();
-    debug!("RPC: {}", &data);
+    trace!("RPC: {}", &data);
 
     let request = http::Request::post(server, vec![data]);
 
@@ -209,7 +209,7 @@ pub fn send_rpc(
         warn!("No UTF8 body");
         EthereumClientError::InvalidUTF8
     })?;
-    debug!("RPC Response: {}", body_str.clone());
+    trace!("RPC Response: {}", body_str.clone());
 
     Ok(String::from(body_str))
 }
@@ -221,13 +221,13 @@ pub fn get_block(
 ) -> Result<EthereumBlock, EthereumClientError> {
     let block_str = encode_block_hex(block_num);
     let block_obj = get_block_object(server, &block_str)?;
-    debug!("eth_starport_address: {:X?}", &eth_starport_address[..]);
+    trace!("eth_starport_address: {:X?}", &eth_starport_address[..]);
     let get_logs_params = vec![serde_json::json!({
         "address": format!("0x{}", ::hex::encode(&eth_starport_address[..])),
         "fromBlock": &block_str,
         "toBlock": &block_str,
     })];
-    debug!("get_logs_params: {:?}", get_logs_params.clone());
+    trace!("get_logs_params: {:?}", get_logs_params.clone());
     let get_logs_response_str: String = send_rpc(server, "eth_getLogs".into(), get_logs_params)?;
     let get_logs_response = deserialize_get_logs_response(&get_logs_response_str)?;
     let event_objects = get_logs_response
@@ -282,7 +282,7 @@ pub fn get_block_object(server: &str, block_num: &str) -> Result<BlockObject, Et
 pub fn get_latest_block_number(server: &str) -> Result<u64, EthereumClientError> {
     let response_str: String = send_rpc(server, "eth_blockNumber".into(), vec![])?;
     let response = deserialize_block_number_response(&response_str)?;
-    debug!("eth_blockNumber response: {:?}", response.result.clone());
+    trace!("eth_blockNumber response: {:?}", response.result.clone());
     parse_u64(Some(response.result.ok_or(EthereumClientError::NoResult)?))
         .ok_or(EthereumClientError::JsonParseError)
 }

our-std/src/lib.rs

@@ -24,7 +24,6 @@ pub use sp_std::rc;
 pub use sp_std::result;
 pub use sp_std::slice;
 pub use sp_std::str;
-pub use sp_std::sync;
 pub use sp_std::vec;
 
 pub mod collections {

our-std/src/log.rs

@@ -6,82 +6,44 @@ pub use log;
 // std or no_std environment. Just use log!("My Log: {}", 5); and
 // things should just magically work.
 
-#[cfg(feature = "std")]
 #[macro_export]
 macro_rules! log {
     ($($arg:tt)*) => {{
         $crate::log::log::info!($($arg)*);
     }}
 }
 
-#[cfg(not(feature = "std"))]
-#[macro_export]
-macro_rules! log {
-    ($($arg:tt)*) => {{
-        sp_runtime::print($crate::alloc::format!($($arg)*).as_str());
-    }}
-}
-
-#[cfg(feature = "std")]
 #[macro_export]
 macro_rules! debug {
     ($($arg:tt)*) => {{
         $crate::log::log::debug!($($arg)*);
     }}
 }
 
-#[cfg(not(feature = "std"))]
 #[macro_export]
-macro_rules! debug {
+macro_rules! trace {
     ($($arg:tt)*) => {{
-        $crate::log!($($arg)*);
+        $crate::log::log::trace!($($arg)*);
     }}
 }
 
-#[cfg(feature = "std")]
 #[macro_export]
 macro_rules! info {
     ($($arg:tt)*) => {{
         $crate::log::log::info!($($arg)*);
     }}
 }
 
-#[cfg(not(feature = "std"))]
-#[macro_export]
-macro_rules! info {
-    ($($arg:tt)*) => {{
-        $crate::log!($($arg)*);
-    }}
-}
-
-#[cfg(feature = "std")]
 #[macro_export]
 macro_rules! warn {
     ($($arg:tt)*) => {{
         $crate::log::log::warn!($($arg)*);
     }}
 }
 
-#[cfg(not(feature = "std"))]
-#[macro_export]
-macro_rules! warn {
-    ($($arg:tt)*) => {{
-        $crate::log!($($arg)*);
-    }}
-}
-
-#[cfg(feature = "std")]
 #[macro_export]
 macro_rules! error {
     ($($arg:tt)*) => {{
         $crate::log::log::error!($($arg)*);
     }}
 }
-
-#[cfg(not(feature = "std"))]
-#[macro_export]
-macro_rules! error {
-    ($($arg:tt)*) => {{
-        $crate::log!($($arg)*);
-    }}
-}

our-std/with_std.rs

@@ -1,3 +1,4 @@
 pub use serde::{Deserialize, Serialize};
 pub use std::fmt::Debug as Debuggable;
+pub use std::sync;
 pub use Debug as RuntimeDebug;

our-std/without_std.rs

@@ -1,4 +1,5 @@
 pub use core::fmt::Debug as Debuggable;
+pub use core::sync;
 pub use serde::{Deserialize, Serialize};
 #[cfg(not(feature = "runtime-debug"))]
 pub use sp_runtime::RuntimeDebug;

pallets/cash/src/core.rs

@@ -23,8 +23,8 @@ use crate::{
         SignersSet, Timestamp, ValidatorKeys,
     },
     AssetBalances, AssetsWithNonZeroBalance, CashIndex, CashPrincipals, CashYield, Config, Event,
-    GlobalCashIndex, IngressionQueue, LastProcessedBlock, Pallet, Starports, SupportedAssets,
-    TotalBorrowAssets, TotalCashPrincipal, TotalSupplyAssets, Validators,
+    FirstBlock, GlobalCashIndex, IngressionQueue, LastProcessedBlock, Pallet, Starports,
+    SupportedAssets, TotalBorrowAssets, TotalCashPrincipal, TotalSupplyAssets, Validators,
 };
 
 use codec::Decode;
@@ -71,9 +71,14 @@ pub fn get_event_queue<T: Config>(chain_id: ChainId) -> Result<ChainBlockEvents,
     Ok(IngressionQueue::get(chain_id).unwrap_or(ChainBlockEvents::empty(chain_id)?))
 }
 
-/// Return the last processed block for the underlying chain, or the initial one for the starport.
+/// Return the last processed block for the underlying chain.
+pub fn get_first_block<T: Config>(chain_id: ChainId) -> Result<ChainBlock, Reason> {
+    FirstBlock::get(chain_id).ok_or(Reason::MissingBlock)
+}
+
+/// Return the last processed block for the underlying chain.
 pub fn get_last_block<T: Config>(chain_id: ChainId) -> Result<ChainBlock, Reason> {
-    LastProcessedBlock::get(chain_id).ok_or(Reason::MissingLastBlock)
+    LastProcessedBlock::get(chain_id).ok_or(Reason::MissingBlock)
 }
 
 /// Return the current total borrow and total supply balances for the asset.

pallets/cash/src/events.rs

@@ -91,12 +91,9 @@ fn fetch_eth_blocks(
 mod tests {
     use crate::events::*;
     use crate::tests::*;
-    use sp_core::offchain::testing;
-    use sp_core::offchain::{OffchainDbExt, OffchainWorkerExt};
 
     #[test]
     fn test_fetch_chain_blocks_eth_returns_proper_blocks() -> Result<(), Reason> {
-        // XXX should this use new_test_ext_with_http_calls?
         let blocks_to_return = vec![
             ethereum_client::EthereumBlock {
                 hash: [1u8; 32],
@@ -116,13 +113,8 @@ mod tests {
         let fetch_to = blocks_to_return[blocks_to_return.len() - 1].number + 1;
         const STARPORT_ADDR: [u8; 20] = [1; 20];
 
-        let (offchain, offchain_state) = testing::TestOffchainExt::new();
-        let mut t = sp_io::TestExternalities::default();
-
-        t.register_extension(OffchainDbExt::new(offchain.clone()));
-        t.register_extension(OffchainWorkerExt::new(offchain));
-
-        gen_mock_responses(offchain_state, blocks_to_return.clone(), STARPORT_ADDR);
+        let calls = gen_mock_calls(&blocks_to_return, STARPORT_ADDR);
+        let (mut t, _, _) = new_test_ext_with_http_calls(calls);
 
         t.execute_with(|| {
             let fetched_blocks = fetch_eth_blocks(fetch_from, fetch_to, &STARPORT_ADDR).unwrap();