diff --git a/masq/src/commands/configuration_command.rs b/masq/src/commands/configuration_command.rs index fea412a37..ab7906cc3 100644 --- a/masq/src/commands/configuration_command.rs +++ b/masq/src/commands/configuration_command.rs @@ -111,6 +111,14 @@ impl ConfigurationCommand { &Self::interpret_option(&configuration.earning_wallet_address_opt), ); dump_parameter_line(stream, "Gas price:", &configuration.gas_price.to_string()); + dump_parameter_line( + stream, + "Max block count:", + &configuration + .max_block_count_opt + .map(|m| m.separate_with_commas()) + .unwrap_or_else(|| "[Unlimited]".to_string()), + ); dump_parameter_line( stream, "Neighborhood mode:", @@ -306,6 +314,7 @@ mod tests { chain_name: "ropsten".to_string(), gas_price: 2345, neighborhood_mode: "standard".to_string(), + max_block_count_opt: None, consuming_wallet_private_key_opt: Some("consuming wallet private key".to_string()), consuming_wallet_address_opt: Some("consuming wallet address".to_string()), earning_wallet_address_opt: Some("earning address".to_string()), @@ -367,6 +376,7 @@ mod tests { |Current schema version: schema version\n\ |Earning wallet address: earning address\n\ |Gas price: 2345\n\ +|Max block count: [Unlimited]\n\ |Neighborhood mode: standard\n\ |Port mapping protocol: PCP\n\ |Start block: 3456\n\ @@ -403,6 +413,7 @@ mod tests { clandestine_port: 1234, chain_name: "mumbai".to_string(), gas_price: 2345, + max_block_count_opt: Some(100_000), neighborhood_mode: "zero-hop".to_string(), consuming_wallet_address_opt: None, consuming_wallet_private_key_opt: None, @@ -463,6 +474,7 @@ mod tests { |Current schema version: schema version\n\ |Earning wallet address: earning wallet\n\ |Gas price: 2345\n\ +|Max block count: 100,000\n\ |Neighborhood mode: zero-hop\n\ |Port mapping protocol: PCP\n\ |Start block: 3456\n\ diff --git a/masq_lib/src/constants.rs b/masq_lib/src/constants.rs index 1371c0b63..7bdcbe1e1 100644 --- a/masq_lib/src/constants.rs +++ b/masq_lib/src/constants.rs @@ -5,7 +5,7 @@ use crate::data_version::DataVersion; use const_format::concatcp; pub const DEFAULT_CHAIN: Chain = Chain::PolyMainnet; -pub const CURRENT_SCHEMA_VERSION: usize = 8; +pub const CURRENT_SCHEMA_VERSION: usize = 9; pub const HIGHEST_RANDOM_CLANDESTINE_PORT: u16 = 9999; pub const HTTP_PORT: u16 = 80; diff --git a/masq_lib/src/messages.rs b/masq_lib/src/messages.rs index b0d6ae15f..e80cf0241 100644 --- a/masq_lib/src/messages.rs +++ b/masq_lib/src/messages.rs @@ -484,6 +484,8 @@ pub struct UiConfigurationResponse { pub earning_wallet_address_opt: Option, #[serde(rename = "gasPrice")] pub gas_price: u64, + #[serde(rename = "maxBlockCount")] + pub max_block_count_opt: Option, #[serde(rename = "neighborhoodMode")] pub neighborhood_mode: String, #[serde(rename = "portMappingProtocol")] diff --git a/multinode_integration_tests/docker/macos/Dockerfile b/multinode_integration_tests/docker/macos/Dockerfile index dd9b3074c..a4a86f156 100644 --- a/multinode_integration_tests/docker/macos/Dockerfile +++ b/multinode_integration_tests/docker/macos/Dockerfile @@ -1,13 +1,13 @@ -FROM rust:stretch +FROM rust:bullseye ARG uid ARG gid -RUN (addgroup substratum --gid $gid || continue) \ - && adduser --disabled-password --uid $uid --gid $gid --home /home/substratum substratum \ - && chown -R $uid:$gid /home/substratum +RUN (addgroup masq --gid $gid || continue) \ + && adduser --disabled-password --uid $uid --gid $gid --home /home/masq masq \ + && chown -R $uid:$gid /home/masq RUN apt-get update && apt-get install -y sudo curl && rustup component add rustfmt clippy \ && cargo install sccache && chown -R $uid:$gid /usr/local/cargo /usr/local/rustup -USER substratum +USER masq diff --git a/node/src/blockchain/blockchain_bridge.rs b/node/src/blockchain/blockchain_bridge.rs index 6eea0ffb9..da7374b58 100644 --- a/node/src/blockchain/blockchain_bridge.rs +++ b/node/src/blockchain/blockchain_bridge.rs @@ -31,17 +31,17 @@ use masq_lib::blockchains::chains::Chain; use masq_lib::logger::Logger; use masq_lib::messages::ScanType; use masq_lib::ui_gateway::NodeFromUiMessage; +use regex::Regex; use std::path::PathBuf; use std::time::SystemTime; use web3::transports::Http; -use web3::types::{TransactionReceipt, H256}; -use web3::Transport; +use web3::types::{BlockNumber, TransactionReceipt, H256}; pub const CRASH_KEY: &str = "BLOCKCHAINBRIDGE"; -pub struct BlockchainBridge { +pub struct BlockchainBridge { consuming_wallet_opt: Option, - blockchain_interface: Box>, + blockchain_interface: Box, logger: Logger, persistent_config: Box, sent_payable_subs_opt: Option>, @@ -331,21 +331,49 @@ impl BlockchainBridge { } fn handle_retrieve_transactions(&mut self, msg: RetrieveTransactions) -> Result<(), String> { - let start_block = match self.persistent_config.start_block() { + let start_block_nbr = match self.persistent_config.start_block() { Ok (sb) => sb, Err (e) => panic! ("Cannot retrieve start block from database; payments to you may not be processed: {:?}", e) }; - let retrieved_transactions = self - .blockchain_interface - .retrieve_transactions(start_block, &msg.recipient); - + let max_block_count = match self.persistent_config.max_block_count() { + Ok(Some(mbc)) => mbc, + _ => u64::MAX, + }; + let end_block = match self.blockchain_interface.get_block_number() { + Ok(eb) => { + if u64::MAX == max_block_count { + BlockNumber::Number(eb) + } else { + BlockNumber::Number(eb.as_u64().min(start_block_nbr + max_block_count).into()) + } + } + Err(e) => { + info!( + self.logger, + "Using 'latest' block number instead of a literal number. {:?}", e + ); + if max_block_count == u64::MAX { + BlockNumber::Latest + } else { + BlockNumber::Number((start_block_nbr + max_block_count).into()) + } + } + }; + let start_block = BlockNumber::Number(start_block_nbr.into()); + let retrieved_transactions = + self.blockchain_interface + .retrieve_transactions(start_block, end_block, &msg.recipient); match retrieved_transactions { Ok(transactions) => { + debug!( + self.logger, + "Write new start block: {}", transactions.new_start_block + ); if let Err(e) = self .persistent_config .set_start_block(transactions.new_start_block) { - panic! ("Cannot set start block in database; payments to you may not be processed: {:?}", e) + panic! ("Cannot set start block {} in database; payments to you may not be processed: {:?}", transactions.new_start_block, e) }; if transactions.transactions.is_empty() { debug!(self.logger, "No new receivable detected"); @@ -361,10 +389,33 @@ impl BlockchainBridge { .expect("Accountant is dead."); Ok(()) } - Err(e) => Err(format!( - "Tried to retrieve received payments but failed: {:?}", - e - )), + Err(e) => { + if let Some(max_block_count) = self.extract_max_block_count(e.clone()) { + debug!(self.logger, "Writing max_block_count({})", max_block_count); + self.persistent_config + .set_max_block_count(Some(max_block_count)) + .map_or_else( + |_| { + warning!(self.logger, "{} update max_block_count to {}. Scheduling next scan with that limit.", e, max_block_count); + Err(format!("{} updated max_block_count to {}. Scheduling next scan with that limit.", e, max_block_count)) + }, + |e| { + warning!(self.logger, "Writing max_block_count failed: {:?}", e); + Err(format!("Writing max_block_count failed: {:?}", e)) + }, + ) + } else { + warning!( + self.logger, + "Attempted to retrieve received payments but failed: {:?}", + e + ); + Err(format!( + "Attempted to retrieve received payments but failed: {:?}", + e + )) + } + } } } @@ -477,6 +528,29 @@ impl BlockchainBridge { .as_ref() .expect("Accountant unbound") } + + pub fn extract_max_block_count(&self, error: BlockchainError) -> Option { + let regex_result = + Regex::new(r".* (max: |allowed for your plan: |is limited to |block range limit \()(?P\d+).*") + .expect("Invalid regex"); + let max_block_count = match error { + BlockchainError::QueryFailed(msg) => match regex_result.captures(msg.as_str()) { + Some(captures) => match captures.name("max_block_count") { + Some(m) => match m.as_str().parse::() { + Ok(value) => Some(value), + Err(_) => None, + }, + _ => None, + }, + None => match msg.as_str() { + "Got invalid response: Expected batch, got single." => Some(1000), + _ => None, + }, + }, + _ => None, + }; + max_block_count + } } #[derive(Debug, PartialEq, Eq, Clone)] @@ -495,7 +569,7 @@ mod tests { use crate::blockchain::bip32::Bip32EncryptionKeyProvider; use crate::blockchain::blockchain_interface::ProcessedPayableFallible::Correct; use crate::blockchain::blockchain_interface::{ - BlockchainError, BlockchainTransaction, RetrievedBlockchainTransactions, + BlockchainError, BlockchainTransaction, LatestBlockNumber, RetrievedBlockchainTransactions, }; use crate::blockchain::test_utils::{make_tx_hash, BlockchainInterfaceMock}; use crate::db_config::persistent_configuration::PersistentConfigError; @@ -1245,10 +1319,14 @@ mod tests { .system_stop_conditions(match_every_type_id!(ScanError)) .start() .recipient(); - let blockchain_interface = BlockchainInterfaceMock::default().retrieve_transactions_result( - Err(BlockchainError::QueryFailed("we have no luck".to_string())), - ); - let persistent_config = PersistentConfigurationMock::new().start_block_result(Ok(5)); // no set_start_block_result: set_start_block() must not be called + let blockchain_interface = BlockchainInterfaceMock::default() + .retrieve_transactions_result(Err(BlockchainError::QueryFailed( + "we have no luck".to_string(), + ))) + .get_block_number_result(LatestBlockNumber::Ok(U64::from(1234u64))); + let persistent_config = PersistentConfigurationMock::new() + .max_block_count_result(Ok(Some(100_000))) + .start_block_result(Ok(5)); // no set_start_block_result: set_start_block() must not be called let mut subject = BlockchainBridge::new( Box::new(blockchain_interface), Box::new(persistent_config), @@ -1273,12 +1351,12 @@ mod tests { &ScanError { scan_type: ScanType::Receivables, response_skeleton_opt: None, - msg: "Tried to retrieve received payments but failed: QueryFailed(\"we have no luck\")".to_string() + msg: "Attempted to retrieve received payments but failed: QueryFailed(\"we have no luck\")".to_string() } ); assert_eq!(recording.len(), 1); TestLogHandler::new().exists_log_containing( - "WARN: BlockchainBridge: Tried to retrieve \ + "WARN: BlockchainBridge: Attempted to retrieve \ received payments but failed: QueryFailed(\"we have no luck\")", ); } @@ -1508,6 +1586,100 @@ mod tests { receipt for '0x000000000000000000000000000000000000000000000000000000000001b2e6' failed due to 'QueryFailed(\"booga\")'"); } + #[test] + fn handle_retrieve_transactions_uses_latest_block_number_upon_get_block_number_error() { + init_test_logging(); + let retrieve_transactions_params_arc = Arc::new(Mutex::new(vec![])); + let system = System::new( + "handle_retrieve_transactions_uses_latest_block_number_upon_get_block_number_error", + ); + let (accountant, _, accountant_recording_arc) = make_recorder(); + let earning_wallet = make_wallet("somewallet"); + let amount = 42; + let amount2 = 55; + let expected_transactions = RetrievedBlockchainTransactions { + new_start_block: 8675309u64, + transactions: vec![ + BlockchainTransaction { + block_number: 7, + from: earning_wallet.clone(), + wei_amount: amount, + }, + BlockchainTransaction { + block_number: 9, + from: earning_wallet.clone(), + wei_amount: amount2, + }, + ], + }; + let blockchain_interface_mock = BlockchainInterfaceMock::default() + .retrieve_transactions_params(&retrieve_transactions_params_arc) + .retrieve_transactions_result(Ok(expected_transactions.clone())) + .get_block_number_result(LatestBlockNumber::Err(BlockchainError::QueryFailed( + "Failed to read the latest block number".to_string(), + ))); + let set_start_block_params_arc = Arc::new(Mutex::new(vec![])); + let persistent_config = PersistentConfigurationMock::new() + .max_block_count_result(Ok(Some(10000u64))) + .start_block_result(Ok(6)) + .set_start_block_params(&set_start_block_params_arc) + .set_start_block_result(Ok(())); + + let subject = BlockchainBridge::new( + Box::new(blockchain_interface_mock), + Box::new(persistent_config), + false, + Some(make_wallet("consuming")), + ); + + let addr = subject.start(); + let subject_subs = BlockchainBridge::make_subs_from(&addr); + let peer_actors = peer_actors_builder().accountant(accountant).build(); + send_bind_message!(subject_subs, peer_actors); + let retrieve_transactions = RetrieveTransactions { + recipient: earning_wallet.clone(), + response_skeleton_opt: Some(ResponseSkeleton { + client_id: 1234, + context_id: 4321, + }), + }; + let before = SystemTime::now(); + + let _ = addr.try_send(retrieve_transactions).unwrap(); + + System::current().stop(); + system.run(); + let after = SystemTime::now(); + let set_start_block_params = set_start_block_params_arc.lock().unwrap(); + assert_eq!(*set_start_block_params, vec![8675309u64]); + let retrieve_transactions_params = retrieve_transactions_params_arc.lock().unwrap(); + assert_eq!( + *retrieve_transactions_params, + vec![( + BlockNumber::Number(6u64.into()), + BlockNumber::Number(10006u64.into()), + earning_wallet + )] + ); + let accountant_received_payment = accountant_recording_arc.lock().unwrap(); + assert_eq!(accountant_received_payment.len(), 1); + let received_payments = accountant_received_payment.get_record::(0); + check_timestamp(before, received_payments.timestamp, after); + assert_eq!( + received_payments, + &ReceivedPayments { + timestamp: received_payments.timestamp, + payments: expected_transactions.transactions, + response_skeleton_opt: Some(ResponseSkeleton { + client_id: 1234, + context_id: 4321 + }), + } + ); + + TestLogHandler::new().exists_log_containing("INFO: BlockchainBridge: Using 'latest' block number instead of a literal number. QueryFailed(\"Failed to read the latest block number\")"); + } + #[test] fn handle_retrieve_transactions_sends_received_payments_back_to_accountant() { let retrieve_transactions_params_arc = Arc::new(Mutex::new(vec![])); @@ -1532,11 +1704,14 @@ mod tests { }, ], }; + let latest_block_number = LatestBlockNumber::Ok(1024u64.into()); let blockchain_interface_mock = BlockchainInterfaceMock::default() .retrieve_transactions_params(&retrieve_transactions_params_arc) - .retrieve_transactions_result(Ok(expected_transactions.clone())); + .retrieve_transactions_result(Ok(expected_transactions.clone())) + .get_block_number_result(latest_block_number); let set_start_block_params_arc = Arc::new(Mutex::new(vec![])); let persistent_config = PersistentConfigurationMock::new() + .max_block_count_result(Ok(Some(10000u64))) .start_block_result(Ok(6)) .set_start_block_params(&set_start_block_params_arc) .set_start_block_result(Ok(())); @@ -1565,9 +1740,16 @@ mod tests { system.run(); let after = SystemTime::now(); let set_start_block_params = set_start_block_params_arc.lock().unwrap(); - assert_eq!(*set_start_block_params, vec![1234]); + assert_eq!(*set_start_block_params, vec![1234u64]); let retrieve_transactions_params = retrieve_transactions_params_arc.lock().unwrap(); - assert_eq!(*retrieve_transactions_params, vec![(6, earning_wallet)]); + assert_eq!( + *retrieve_transactions_params, + vec![( + BlockNumber::Number(6u64.into()), + BlockNumber::Number(1024u64.into()), + earning_wallet + )] + ); let accountant_received_payment = accountant_recording_arc.lock().unwrap(); assert_eq!(accountant_received_payment.len(), 1); let received_payments = accountant_received_payment.get_record::(0); @@ -1592,9 +1774,11 @@ mod tests { .retrieve_transactions_result(Ok(RetrievedBlockchainTransactions { new_start_block: 7, transactions: vec![], - })); + })) + .get_block_number_result(Ok(0u64.into())); let set_start_block_params_arc = Arc::new(Mutex::new(vec![])); let persistent_config = PersistentConfigurationMock::new() + .max_block_count_result(Ok(Some(10000u64))) .start_block_result(Ok(6)) .set_start_block_params(&set_start_block_params_arc) .set_start_block_result(Ok(())); @@ -1651,10 +1835,12 @@ mod tests { expected = "Cannot retrieve start block from database; payments to you may not be processed: TransactionError" )] fn handle_retrieve_transactions_panics_if_start_block_cannot_be_read() { + let blockchain_interface = + BlockchainInterfaceMock::default().get_block_number_result(Ok(0u64.into())); let persistent_config = PersistentConfigurationMock::new() .start_block_result(Err(PersistentConfigError::TransactionError)); let mut subject = BlockchainBridge::new( - Box::new(BlockchainInterfaceMock::default()), + Box::new(blockchain_interface), Box::new(persistent_config), false, None, //not needed in this test @@ -1669,22 +1855,23 @@ mod tests { #[test] #[should_panic( - expected = "Cannot set start block in database; payments to you may not be processed: TransactionError" + expected = "Cannot set start block 1234 in database; payments to you may not be processed: TransactionError" )] fn handle_retrieve_transactions_panics_if_start_block_cannot_be_written() { let persistent_config = PersistentConfigurationMock::new() .start_block_result(Ok(1234)) - .set_start_block_result(Err(PersistentConfigError::TransactionError)); - let blockchain_interface = BlockchainInterfaceMock::default().retrieve_transactions_result( - Ok(RetrievedBlockchainTransactions { + .set_start_block_result(Err(PersistentConfigError::TransactionError)) + .max_block_count_result(Ok(Some(10000u64))); + let blockchain_interface = BlockchainInterfaceMock::default() + .get_block_number_result(Ok(0u64.into())) + .retrieve_transactions_result(Ok(RetrievedBlockchainTransactions { new_start_block: 1234, transactions: vec![BlockchainTransaction { block_number: 1000, from: make_wallet("somewallet"), wei_amount: 2345, }], - }), - ); + })); let mut subject = BlockchainBridge::new( Box::new(blockchain_interface), Box::new(persistent_config), @@ -1842,6 +2029,127 @@ mod tests { prove_that_crash_request_handler_is_hooked_up(subject, CRASH_KEY); } + #[test] + fn extract_max_block_range_from_error_response() { + let result = BlockchainError::QueryFailed("RPC error: Error { code: ServerError(-32005), message: \"eth_getLogs block range too large, range: 33636, max: 3500\", data: None }".to_string()); + let subject = BlockchainBridge::new( + Box::new(BlockchainInterfaceMock::default()), + Box::new(PersistentConfigurationMock::default()), + false, + None, + ); + let max_block_count = subject.extract_max_block_count(result); + + assert_eq!(Some(3500u64), max_block_count); + } + + #[test] + fn extract_max_block_range_from_pokt_error_response() { + let result = BlockchainError::QueryFailed("Rpc(Error { code: ServerError(-32001), message: \"Relay request failed validation: invalid relay request: eth_getLogs block range limit (100000 blocks) exceeded\", data: None })".to_string()); + let subject = BlockchainBridge::new( + Box::new(BlockchainInterfaceMock::default()), + Box::new(PersistentConfigurationMock::default()), + false, + None, + ); + let max_block_count = subject.extract_max_block_count(result); + + assert_eq!(Some(100000u64), max_block_count); + } + /* + POKT (Polygon mainnet and mumbai) + {"jsonrpc":"2.0","id":7,"error":{"message":"You cannot query logs for more than 100000 blocks at once.","code":-32064}} + */ + /* + Ankr + {"jsonrpc":"2.0","error":{"code":-32600,"message":"block range is too wide"},"id":null}% + */ + #[test] + fn extract_max_block_range_for_ankr_error_response() { + let result = BlockchainError::QueryFailed("RPC error: Error { code: ServerError(-32600), message: \"block range is too wide\", data: None }".to_string()); + let subject = BlockchainBridge::new( + Box::new(BlockchainInterfaceMock::default()), + Box::new(PersistentConfigurationMock::default()), + false, + None, + ); + let max_block_count = subject.extract_max_block_count(result); + + assert_eq!(None, max_block_count); + } + + /* + MaticVigil + [{"error":{"message":"Blockheight too far in the past. Check params passed to eth_getLogs or eth_call requests.Range of blocks allowed for your plan: 1000","code":-32005},"jsonrpc":"2.0","id":7},{"error":{"message":"Blockheight too far in the past. Check params passed to eth_getLogs or eth_call requests.Range of blocks allowed for your plan: 1000","code":-32005},"jsonrpc":"2.0","id":8}]% + */ + #[test] + fn extract_max_block_range_for_matic_vigil_error_response() { + let result = BlockchainError::QueryFailed("RPC error: Error { code: ServerError(-32005), message: \"Blockheight too far in the past. Check params passed to eth_getLogs or eth_call requests.Range of blocks allowed for your plan: 1000\", data: None }".to_string()); + let subject = BlockchainBridge::new( + Box::new(BlockchainInterfaceMock::default()), + Box::new(PersistentConfigurationMock::default()), + false, + None, + ); + let max_block_count = subject.extract_max_block_count(result); + + assert_eq!(Some(1000), max_block_count); + } + + /* + Blockpi + [{"jsonrpc":"2.0","id":7,"result":"0x21db466"},{"jsonrpc":"2.0","id":8,"error":{"code":-32602,"message":"eth_getLogs is limited to 1024 block range. Please check the parameter requirements at https://docs.blockpi.io/documentations/api-reference"}}] + */ + #[test] + fn extract_max_block_range_for_blockpi_error_response() { + let result = BlockchainError::QueryFailed("RPC error: Error { code: ServerError(-32005), message: \"eth_getLogs is limited to 1024 block range. Please check the parameter requirements at https://docs.blockpi.io/documentations/api-reference\", data: None }".to_string()); + let subject = BlockchainBridge::new( + Box::new(BlockchainInterfaceMock::default()), + Box::new(PersistentConfigurationMock::default()), + false, + None, + ); + let max_block_count = subject.extract_max_block_count(result); + + assert_eq!(Some(1024), max_block_count); + } + + /* + blastapi - completely rejected call on Public endpoint as won't handle eth_getLogs method on public API + [{"jsonrpc":"2.0","id":2,"error":{"code":-32601,"message":"Method not found","data":{"method":""}}},{"jsonrpc":"2.0","id":1,"error":{"code":-32600,"message":"Invalid Request","data":{"message":"Cancelled due to validation errors in batch request"}}}] (edited) + [8:50 AM] + */ + + #[test] + fn extract_max_block_range_for_blastapi_error_response() { + let result = BlockchainError::QueryFailed("RPC error: Error { code: ServerError(-32601), message: \"Method not found\", data: \"'eth_getLogs' is not available on our public API. Head over to https://docs.blastapi.io/blast-documentation/tutorials-and-guides/using-blast-to-get-a-blockchain-endpoint for more information\" }".to_string()); + let subject = BlockchainBridge::new( + Box::new(BlockchainInterfaceMock::default()), + Box::new(PersistentConfigurationMock::default()), + false, + None, + ); + let max_block_count = subject.extract_max_block_count(result); + + assert_eq!(None, max_block_count); + } + + #[test] + fn extract_max_block_range_for_expected_batch_got_single_error_response() { + let result = BlockchainError::QueryFailed( + "Got invalid response: Expected batch, got single.".to_string(), + ); + let subject = BlockchainBridge::new( + Box::new(BlockchainInterfaceMock::default()), + Box::new(PersistentConfigurationMock::default()), + false, + None, + ); + let max_block_count = subject.extract_max_block_count(result); + + assert_eq!(Some(1000), max_block_count); + } + #[test] fn make_connections_implements_panic_on_migration() { let data_dir = ensure_node_home_directory_exists( diff --git a/node/src/blockchain/blockchain_interface.rs b/node/src/blockchain/blockchain_interface.rs index d8bf0d411..5616a70a6 100644 --- a/node/src/blockchain/blockchain_interface.rs +++ b/node/src/blockchain/blockchain_interface.rs @@ -9,8 +9,10 @@ use crate::blockchain::blockchain_interface::BlockchainError::{ }; use crate::sub_lib::wallet::Wallet; use actix::{Message, Recipient}; -use futures::{future, Future}; +use ethereum_types::U64; +use futures::Future; use indoc::indoc; +use itertools::fold; use itertools::Either::{Left, Right}; use masq_lib::blockchains::chains::{Chain, ChainFamily}; use masq_lib::logger::Logger; @@ -23,12 +25,12 @@ use std::iter::once; use thousands::Separable; use variant_count::VariantCount; use web3::contract::{Contract, Options}; -use web3::transports::{Batch, EventLoopHandle, Http}; +use web3::transports::{Batch, EventLoopHandle}; use web3::types::{ Address, BlockNumber, Bytes, FilterBuilder, Log, SignedTransaction, TransactionParameters, TransactionReceipt, H160, H256, U256, }; -use web3::{BatchTransport, Error, Transport, Web3}; +use web3::{BatchTransport, Error, Web3}; pub const REQUESTS_IN_PARALLEL: usize = 1; @@ -69,11 +71,11 @@ pub struct BlockchainTransaction { pub wei_amount: u128, } -impl fmt::Display for BlockchainTransaction { +impl Display for BlockchainTransaction { fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), fmt::Error> { write!( f, - "{}gw from {} ({})", + "{}wei from {} ({:?})", self.wei_amount, self.from, self.block_number ) } @@ -112,6 +114,7 @@ pub type ResultForBalance = BlockchainResult; pub type ResultForBothBalances = BlockchainResult<(web3::types::U256, web3::types::U256)>; pub type ResultForNonce = BlockchainResult; pub type ResultForReceipt = BlockchainResult>; +pub type LatestBlockNumber = BlockchainResult; #[derive(Clone, Debug, PartialEq, Eq, VariantCount)] pub enum PayableTransactionError { @@ -163,12 +166,13 @@ pub struct RetrievedBlockchainTransactions { pub transactions: Vec, } -pub trait BlockchainInterface { +pub trait BlockchainInterface { fn contract_address(&self) -> Address; fn retrieve_transactions( &self, - start_block: u64, + start_block: BlockNumber, + end_block: BlockNumber, recipient: &Wallet, ) -> Result; @@ -181,6 +185,8 @@ pub trait BlockchainInterface { accounts: &[PayableAccount], ) -> Result, PayableTransactionError>; + fn get_block_number(&self) -> LatestBlockNumber; + fn get_transaction_fee_balance(&self, address: &Wallet) -> ResultForBalance; fn get_token_balance(&self, address: &Wallet) -> ResultForBalance; @@ -208,7 +214,8 @@ impl BlockchainInterface for BlockchainInterfaceNull { fn retrieve_transactions( &self, - _start_block: u64, + _start_block: BlockNumber, + _end_block: BlockNumber, _recipient: &Wallet, ) -> Result { self.handle_uninitialized_interface("retrieve transactions") @@ -225,6 +232,12 @@ impl BlockchainInterface for BlockchainInterfaceNull { self.handle_uninitialized_interface("pay payables") } + fn get_block_number(&self) -> LatestBlockNumber { + let msg = "Can't get latest block number clandestinely yet"; + error!(self.logger, "{}", msg); + Err(BlockchainError::QueryFailed(msg.to_string())) + } + fn get_transaction_fee_balance(&self, _address: &Wallet) -> ResultForBalance { self.handle_uninitialized_interface("get transaction fee balance") } @@ -275,27 +288,30 @@ impl BlockchainInterfaceNull { } } -pub struct BlockchainInterfaceWeb3 { +pub struct BlockchainInterfaceWeb3 +where + T: 'static + BatchTransport + Debug, +{ logger: Logger, chain: Chain, // This must not be dropped for Web3 requests to be completed _event_loop_handle: EventLoopHandle, web3: Web3, - batch_web3: Web3>, + web3_batch: Web3>, batch_payable_tools: Box>, contract: Contract, } const GWEI: U256 = U256([1_000_000_000u64, 0, 0, 0]); -pub fn to_wei(gwub: u64) -> U256 { - let subgwei = U256::from(gwub); - subgwei.full_mul(GWEI).try_into().expect("Internal Error") +pub fn to_wei(gwei: u64) -> U256 { + let result = U256::from(gwei); + result.full_mul(GWEI).try_into().expect("Internal Error") } impl BlockchainInterface for BlockchainInterfaceWeb3 where - T: BatchTransport + Debug + 'static, + T: 'static + BatchTransport + Debug, { fn contract_address(&self) -> Address { self.chain.rec().contract @@ -303,21 +319,23 @@ where fn retrieve_transactions( &self, - start_block: u64, + start_block: BlockNumber, + end_block: BlockNumber, recipient: &Wallet, ) -> Result { debug!( self.logger, - "Retrieving transactions from start block: {} for: {} chain_id: {} contract: {:#x}", + "Retrieving transactions from start block: {:?} to end block: {:?} for: {} chain_id: {} contract: {:#x}", start_block, + end_block, recipient, self.chain.rec().num_chain_id, self.contract_address() ); let filter = FilterBuilder::default() .address(vec![self.contract_address()]) - .from_block(BlockNumber::Number(ethereum_types::U64::from(start_block))) - .to_block(BlockNumber::Latest) + .from_block(start_block) + .to_block(end_block) .topics( Some(vec![TRANSACTION_LITERAL]), None, @@ -326,13 +344,39 @@ where ) .build(); - let log_request = self.web3.eth().logs(filter); + let fallback_start_block_number = match end_block { + BlockNumber::Number(eb) => eb.as_u64(), + _ => { + if let BlockNumber::Number(start_block_number) = start_block { + start_block_number.as_u64() + 1u64 + } else { + panic!("start_block of Latest, Earliest, and Pending are not supported"); + } + } + }; + let block_request = self.web3_batch.eth().block_number(); + let log_request = self.web3_batch.eth().logs(filter); + let logger = self.logger.clone(); - log_request - .then(|logs| { - debug!(logger, "Transaction retrieval completed: {:?}", logs); - future::result::(match logs { + match self.web3_batch.transport().submit_batch().wait() { + Ok(_) => { + let response_block_number = match block_request.wait() { + Ok(block_nbr) => { + debug!(logger, "Latest block number: {}", block_nbr.as_u64()); + block_nbr.as_u64() + } + Err(_) => { + debug!( + logger, + "Using fallback block number: {}", fallback_start_block_number + ); + fallback_start_block_number + } + }; + + match log_request.wait() { Ok(logs) => { + let logs_len = logs.len(); if logs .iter() .any(|log| log.topics.len() < 2 || log.data.0.len() > 32) @@ -344,45 +388,44 @@ where ); Err(BlockchainError::InvalidResponse) } else { - let transactions: Vec = logs - .iter() - .filter_map(|log: &Log| match log.block_number { - Some(block_number) => { - let amount: U256 = U256::from(log.data.0.as_slice()); - let wei_amount_result = u128::try_from(amount); - wei_amount_result.ok().map(|wei_amount| { - BlockchainTransaction { - block_number: u64::try_from(block_number) - .expect("Internal Error"), - from: Wallet::from(log.topics[1]), - wei_amount, - } - }) - } - None => None, - }) - .collect(); + let transactions: Vec = + self.extract_transactions_from_logs(logs); debug!(logger, "Retrieved transactions: {:?}", transactions); + if transactions.is_empty() && logs_len != transactions.len() { + warning!( + logger, + "Retrieving transactions: logs: {}, transactions: {}", + logs_len, + transactions.len() + ) + } // Get the largest transaction block number, unless there are no - // transactions, in which case use start_block. - let last_transaction_block = - transactions.iter().fold(start_block, |so_far, elem| { - if elem.block_number > so_far { - elem.block_number - } else { - so_far - } - }); + // transactions, in which case use end_block, unless get_latest_block() + // was not successful. + let transaction_max_block_number = self + .find_largest_transaction_block_number( + response_block_number, + &transactions, + ); + debug!( + logger, + "Discovered transaction max block nbr: {}", + transaction_max_block_number + ); Ok(RetrievedBlockchainTransactions { - new_start_block: last_transaction_block + 1, + new_start_block: 1u64 + transaction_max_block_number, transactions, }) } } - Err(e) => Err(BlockchainError::QueryFailed(e.to_string())), - }) - }) - .wait() + Err(e) => { + error!(self.logger, "Retrieving transactions: {:?}", e); + Err(BlockchainError::QueryFailed(e.to_string())) + } + } + } + Err(e) => Err(BlockchainError::QueryFailed(e.to_string())), + } } fn send_payables_within_batch( @@ -422,7 +465,7 @@ where self.transmission_log(accounts, gas_price) ); - match self.batch_payable_tools.submit_batch(&self.batch_web3) { + match self.batch_payable_tools.submit_batch(&self.web3_batch) { Ok(responses) => Ok(Self::merged_output_data( responses, hashes_and_paid_amounts, @@ -432,6 +475,14 @@ where } } + fn get_block_number(&self) -> LatestBlockNumber { + self.web3 + .eth() + .block_number() + .map_err(|e| BlockchainError::QueryFailed(e.to_string())) + .wait() + } + fn get_transaction_fee_balance(&self, wallet: &Wallet) -> ResultForBalance { self.web3 .eth() @@ -487,11 +538,11 @@ type HashAndAmountResult = Result, PayableTransactionError>; impl BlockchainInterfaceWeb3 where - T: BatchTransport + Debug + 'static, + T: 'static + BatchTransport + Debug, { pub fn new(transport: T, event_loop_handle: EventLoopHandle, chain: Chain) -> Self { let web3 = Web3::new(transport.clone()); - let batch_web3 = Web3::new(Batch::new(transport)); + let web3_batch = Web3::new(Batch::new(transport)); let batch_payable_tools = Box::new(BatchPayableToolsReal::::default()); let contract = Contract::from_json(web3.eth(), chain.rec().contract, CONTRACT_ABI.as_bytes()) @@ -502,7 +553,7 @@ where chain, _event_loop_handle: event_loop_handle, web3, - batch_web3, + web3_batch, batch_payable_tools, contract, } @@ -646,7 +697,7 @@ where let signed_tx = self.sign_transaction(recipient, consuming_wallet, amount, nonce, gas_price)?; self.batch_payable_tools - .append_transaction_to_batch(signed_tx.raw_transaction, &self.batch_web3); + .append_transaction_to_batch(signed_tx.raw_transaction, &self.web3_batch); Ok(signed_tx.transaction_hash) } @@ -695,7 +746,7 @@ where }; self.batch_payable_tools - .sign_transaction(transaction_parameters, &self.batch_web3, &key) + .sign_transaction(transaction_parameters, &self.web3_batch, &key) .map_err(|e| PayableTransactionError::Signing(e.to_string())) } @@ -737,6 +788,36 @@ where } } + fn extract_transactions_from_logs(&self, logs: Vec) -> Vec { + logs.iter() + .filter_map(|log: &Log| match log.block_number { + None => None, + Some(block_number) => { + let wei_amount = U256::from(log.data.0.as_slice()).as_u128(); + Some(BlockchainTransaction { + block_number: block_number.as_u64(), + from: Wallet::from(log.topics[1]), + wei_amount, + }) + } + }) + .collect() + } + + fn find_largest_transaction_block_number( + &self, + response_block_number: u64, + transactions: &Vec, + ) -> u64 { + if transactions.is_empty() { + response_block_number + } else { + fold(transactions, response_block_number, |a, b| { + a.max(b.block_number) + }) + } + } + #[cfg(test)] fn web3(&self) -> &Web3 { &self.web3 @@ -758,9 +839,9 @@ mod tests { BatchPayableToolsMock, TestTransport, }; use crate::sub_lib::wallet::Wallet; - use crate::test_utils::make_paying_wallet; use crate::test_utils::recorder::{make_recorder, Recorder}; use crate::test_utils::unshared_test_utils::decode_hex; + use crate::test_utils::{assert_string_contains, make_paying_wallet}; use crate::test_utils::{make_wallet, TestRawTransaction}; use actix::{Actor, System}; use crossbeam_channel::{unbounded, Receiver}; @@ -772,8 +853,7 @@ mod tests { use masq_lib::test_utils::utils::TEST_DEFAULT_CHAIN; use masq_lib::utils::{find_free_port, slice_of_strs_to_vec_of_strings}; use serde_derive::Deserialize; - use serde_json::json; - use serde_json::Value; + use serde_json::{json, Value}; use simple_server::{Request, Server}; use std::io::Write; use std::net::{IpAddr, Ipv4Addr, SocketAddr, TcpStream}; @@ -895,40 +975,48 @@ mod tests { let port = find_free_port(); let test_server = TestServer::start( port, - vec![br#"{"jsonrpc":"2.0","id":3,"result":[]}"#.to_vec()], + vec![br#"[{"jsonrpc":"2.0","id":2,"result":"0x400"},{"jsonrpc":"2.0","id":3,"result":[]}]"#.to_vec()], ); - let (event_loop_handle, transport) = Http::with_max_parallel( - &format!("http://{}:{}", &Ipv4Addr::LOCALHOST.to_string(), port), + &format!("http://{}:{}", &Ipv4Addr::LOCALHOST, port), REQUESTS_IN_PARALLEL, ) .unwrap(); let subject = BlockchainInterfaceWeb3::new(transport, event_loop_handle, TEST_DEFAULT_CHAIN); + let end_block_nbr = 1024u64; let result = subject .retrieve_transactions( - 42, - &Wallet::from_str("0x3f69f9efd4f2592fd70be8c32ecd9dce71c472fc").unwrap(), + BlockNumber::Number(42u64.into()), + BlockNumber::Number(end_block_nbr.into()), + &Wallet::from_str(&to).unwrap(), ) .unwrap(); let requests = test_server.requests_so_far(); - let bodies: Vec = requests + + let bodies: Vec = requests .into_iter() .map(|request| serde_json::from_slice(&request.body()).unwrap()) + .map(|b: Value| serde_json::to_string(&b).unwrap()) .collect(); - assert_eq!( - format!("\"0x000000000000000000000000{}\"", &to[2..]), - bodies[0]["params"][0]["topics"][2].to_string(), - ); + let expected_body_prefix = r#"[{"id":0,"jsonrpc":"2.0","method":"eth_blockNumber","params":[]},{"id":1,"jsonrpc":"2.0","method":"eth_getLogs","params":[{"address":"0x384dec25e03f94931767ce4c3556168468ba24c3","fromBlock":"0x2a","toBlock":"0x400","topics":["0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",null,"0x000000000000000000000000"#; + let expected_body_suffix = r#""]}]}]"#; + let expected_body = format!( + "{}{}{}", + expected_body_prefix, + &to[2..], + expected_body_suffix + ); + assert_eq!(bodies, vec!(expected_body)); assert_eq!( result, RetrievedBlockchainTransactions { - new_start_block: 42 + 1, + new_start_block: 1 + end_block_nbr, transactions: vec![] } - ) + ); } #[test] @@ -937,7 +1025,7 @@ mod tests { let port = find_free_port(); #[rustfmt::skip] let test_server = TestServer::start (port, vec![ - br#"{ + br#"[{"jsonrpc":"2.0","id":2,"result":"0x400"},{ "jsonrpc":"2.0", "id":3, "result":[ @@ -972,49 +1060,57 @@ mod tests { "transactionIndex":"0x0" } ] - }"#.to_vec(), + }]"#.to_vec(), ]); let (event_loop_handle, transport) = Http::with_max_parallel( - &format!("http://{}:{}", &Ipv4Addr::LOCALHOST.to_string(), port), + &format!("http://{}:{}", &Ipv4Addr::LOCALHOST, port), REQUESTS_IN_PARALLEL, ) .unwrap(); let subject = BlockchainInterfaceWeb3::new(transport, event_loop_handle, TEST_DEFAULT_CHAIN); + let end_block_nbr = 1024u64; let result = subject .retrieve_transactions( - 42, - &Wallet::from_str("0x3f69f9efd4f2592fd70be8c32ecd9dce71c472fc").unwrap(), + BlockNumber::Number(42u64.into()), + BlockNumber::Number(end_block_nbr.into()), + &Wallet::from_str(&to).unwrap(), ) .unwrap(); let requests = test_server.requests_so_far(); - let bodies: Vec = requests + let bodies: Vec = requests .into_iter() .map(|request| serde_json::from_slice(&request.body()).unwrap()) + .map(|b: Value| serde_json::to_string(&b).unwrap()) .collect(); - assert_eq!( - format!("\"0x000000000000000000000000{}\"", &to[2..]), - bodies[0]["params"][0]["topics"][2].to_string(), - ); + let expected_body_prefix = r#"[{"id":0,"jsonrpc":"2.0","method":"eth_blockNumber","params":[]},{"id":1,"jsonrpc":"2.0","method":"eth_getLogs","params":[{"address":"0x384dec25e03f94931767ce4c3556168468ba24c3","fromBlock":"0x2a","toBlock":"0x400","topics":["0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",null,"0x000000000000000000000000"#; + let expected_body_suffix = r#""]}]}]"#; + let expected_body = format!( + "{}{}{}", + expected_body_prefix, + &to[2..], + expected_body_suffix + ); + assert_eq!(bodies, vec!(expected_body)); assert_eq!( result, RetrievedBlockchainTransactions { - new_start_block: 0x4be663 + 1, + new_start_block: 0x4be664, transactions: vec![ BlockchainTransaction { block_number: 0x4be663, from: Wallet::from_str("0x3ab28ecedea6cdb6feed398e93ae8c7b316b1182") .unwrap(), - wei_amount: 4_503_599_627_370_496, + wei_amount: 4_503_599_627_370_496u128, }, BlockchainTransaction { block_number: 0x4be662, from: Wallet::from_str("0x3f69f9efd4f2592fd70be8c32ecd9dce71c472fc") .unwrap(), - wei_amount: 4_503_599_627_370_496, + wei_amount: 4_503_599_627_370_496u128, }, ] } @@ -1034,8 +1130,11 @@ mod tests { let subject = BlockchainInterfaceWeb3::new(transport, event_loop_handle, TEST_DEFAULT_CHAIN); - let result = subject - .retrieve_transactions(42, &Wallet::new("0x3f69f9efd4f2592fd70beecd9dce71c472fc")); + let result = subject.retrieve_transactions( + BlockNumber::Number(42u64.into()), + BlockNumber::Latest, + &Wallet::new("0x3f69f9efd4f2592fd70beecd9dce71c472fc"), + ); assert_eq!( result.expect_err("Expected an Err, got Ok"), @@ -1048,10 +1147,10 @@ mod tests { ) { let port = find_free_port(); let _test_server = TestServer::start (port, vec![ - br#"{"jsonrpc":"2.0","id":3,"result":[{"address":"0xcd6c588e005032dd882cd43bf53a32129be81302","blockHash":"0x1a24b9169cbaec3f6effa1f600b70c7ab9e8e86db44062b49132a4415d26732a","blockNumber":"0x4be663","data":"0x0000000000000000000000000000000000000000000000056bc75e2d63100000","logIndex":"0x0","removed":false,"topics":["0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"],"transactionHash":"0x955cec6ac4f832911ab894ce16aa22c3003f46deff3f7165b32700d2f5ff0681","transactionIndex":"0x0"}]}"#.to_vec() + br#"[{"jsonrpc":"2.0","id":2,"result":"0x400"},{"jsonrpc":"2.0","id":3,"result":[{"address":"0xcd6c588e005032dd882cd43bf53a32129be81302","blockHash":"0x1a24b9169cbaec3f6effa1f600b70c7ab9e8e86db44062b49132a4415d26732a","blockNumber":"0x4be663","data":"0x0000000000000000000000000000000000000000000000056bc75e2d63100000","logIndex":"0x0","removed":false,"topics":["0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"],"transactionHash":"0x955cec6ac4f832911ab894ce16aa22c3003f46deff3f7165b32700d2f5ff0681","transactionIndex":"0x0"}]}]"#.to_vec() ]); let (event_loop_handle, transport) = Http::with_max_parallel( - &format!("http://{}:{}", &Ipv4Addr::LOCALHOST.to_string(), port), + &format!("http://{}:{}", &Ipv4Addr::LOCALHOST, port), REQUESTS_IN_PARALLEL, ) .unwrap(); @@ -1059,7 +1158,8 @@ mod tests { BlockchainInterfaceWeb3::new(transport, event_loop_handle, TEST_DEFAULT_CHAIN); let result = subject.retrieve_transactions( - 42, + BlockNumber::Number(42u64.into()), + BlockNumber::Latest, &Wallet::from_str("0x3f69f9efd4f2592fd70be8c32ecd9dce71c472fc").unwrap(), ); @@ -1074,11 +1174,11 @@ mod tests { ) { let port = find_free_port(); let _test_server = TestServer::start(port, vec![ - br#"{"jsonrpc":"2.0","id":3,"result":[{"address":"0xcd6c588e005032dd882cd43bf53a32129be81302","blockHash":"0x1a24b9169cbaec3f6effa1f600b70c7ab9e8e86db44062b49132a4415d26732a","blockNumber":"0x4be663","data":"0x0000000000000000000000000000000000000000000000056bc75e2d6310000001","logIndex":"0x0","removed":false,"topics":["0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef","0x0000000000000000000000003f69f9efd4f2592fd70be8c32ecd9dce71c472fc","0x000000000000000000000000adc1853c7859369639eb414b6342b36288fe6092"],"transactionHash":"0x955cec6ac4f832911ab894ce16aa22c3003f46deff3f7165b32700d2f5ff0681","transactionIndex":"0x0"}]}"#.to_vec() + br#"[{"jsonrpc":"2.0","id":2,"result":"0x400"},{"jsonrpc":"2.0","id":3,"result":[{"address":"0xcd6c588e005032dd882cd43bf53a32129be81302","blockHash":"0x1a24b9169cbaec3f6effa1f600b70c7ab9e8e86db44062b49132a4415d26732a","blockNumber":"0x4be663","data":"0x0000000000000000000000000000000000000000000000056bc75e2d6310000001","logIndex":"0x0","removed":false,"topics":["0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef","0x0000000000000000000000003f69f9efd4f2592fd70be8c32ecd9dce71c472fc","0x000000000000000000000000adc1853c7859369639eb414b6342b36288fe6092"],"transactionHash":"0x955cec6ac4f832911ab894ce16aa22c3003f46deff3f7165b32700d2f5ff0681","transactionIndex":"0x0"}]}]"#.to_vec() ]); let (event_loop_handle, transport) = Http::with_max_parallel( - &format!("http://{}:{}", &Ipv4Addr::LOCALHOST.to_string(), port), + &format!("http://{}:{}", &Ipv4Addr::LOCALHOST, port), REQUESTS_IN_PARALLEL, ) .unwrap(); @@ -1087,7 +1187,8 @@ mod tests { BlockchainInterfaceWeb3::new(transport, event_loop_handle, TEST_DEFAULT_CHAIN); let result = subject.retrieve_transactions( - 42, + BlockNumber::Number(42u64.into()), + BlockNumber::Latest, &Wallet::from_str("0x3f69f9efd4f2592fd70be8c32ecd9dce71c472fc").unwrap(), ); @@ -1099,11 +1200,48 @@ mod tests { ) { let port = find_free_port(); let _test_server = TestServer::start (port, vec![ - br#"{"jsonrpc":"2.0","id":3,"result":[{"address":"0xcd6c588e005032dd882cd43bf53a32129be81302","blockHash":"0x1a24b9169cbaec3f6effa1f600b70c7ab9e8e86db44062b49132a4415d26732a","data":"0x0000000000000000000000000000000000000000000000000010000000000000","logIndex":"0x0","removed":false,"topics":["0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef","0x0000000000000000000000003f69f9efd4f2592fd70be8c32ecd9dce71c472fc","0x000000000000000000000000adc1853c7859369639eb414b6342b36288fe6092"],"transactionHash":"0x955cec6ac4f832911ab894ce16aa22c3003f46deff3f7165b32700d2f5ff0681","transactionIndex":"0x0"}]}"#.to_vec() + br#"[{"jsonrpc":"2.0","id":1,"result":"0x400"},{"jsonrpc":"2.0","id":2,"result":[{"address":"0xcd6c588e005032dd882cd43bf53a32129be81302","blockHash":"0x1a24b9169cbaec3f6effa1f600b70c7ab9e8e86db44062b49132a4415d26732a","data":"0x0000000000000000000000000000000000000000000000000010000000000000","logIndex":"0x0","removed":false,"topics":["0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef","0x0000000000000000000000003f69f9efd4f2592fd70be8c32ecd9dce71c472fc","0x000000000000000000000000adc1853c7859369639eb414b6342b36288fe6092"],"transactionHash":"0x955cec6ac4f832911ab894ce16aa22c3003f46deff3f7165b32700d2f5ff0681","transactionIndex":"0x0"}]}]"#.to_vec() + ]); + init_test_logging(); + let (event_loop_handle, transport) = Http::with_max_parallel( + &format!("http://{}:{}", &Ipv4Addr::LOCALHOST, port), + REQUESTS_IN_PARALLEL, + ) + .unwrap(); + + let end_block_nbr = 1024u64; + let subject = + BlockchainInterfaceWeb3::new(transport, event_loop_handle, TEST_DEFAULT_CHAIN); + + let result = subject.retrieve_transactions( + BlockNumber::Number(42u64.into()), + BlockNumber::Number(end_block_nbr.into()), + &Wallet::from_str("0x3f69f9efd4f2592fd70be8c32ecd9dce71c472fc").unwrap(), + ); + + assert_eq!( + result, + Ok(RetrievedBlockchainTransactions { + new_start_block: 1 + end_block_nbr, + transactions: vec![] + }) + ); + let test_log_handler = TestLogHandler::new(); + test_log_handler.exists_log_containing( + "WARN: BlockchainInterface: Retrieving transactions: logs: 1, transactions: 0", + ); + } + + #[test] + fn blockchain_interface_non_clandestine_retrieve_transactions_uses_block_number_latest_as_fallback_start_block_plus_one( + ) { + let port = find_free_port(); + let _test_server = TestServer::start (port, vec![ + br#"[{"jsonrpc":"2.0","id":1,"result":"error"},{"jsonrpc":"2.0","id":2,"result":[{"address":"0xcd6c588e005032dd882cd43bf53a32129be81302","blockHash":"0x1a24b9169cbaec3f6effa1f600b70c7ab9e8e86db44062b49132a4415d26732a","data":"0x0000000000000000000000000000000000000000000000000010000000000000","logIndex":"0x0","removed":false,"topics":["0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef","0x0000000000000000000000003f69f9efd4f2592fd70be8c32ecd9dce71c472fc","0x000000000000000000000000adc1853c7859369639eb414b6342b36288fe6092"],"transactionHash":"0x955cec6ac4f832911ab894ce16aa22c3003f46deff3f7165b32700d2f5ff0681","transactionIndex":"0x0"}]}]"#.to_vec() ]); let (event_loop_handle, transport) = Http::with_max_parallel( - &format!("http://{}:{}", &Ipv4Addr::LOCALHOST.to_string(), port), + &format!("http://{}:{}", &Ipv4Addr::LOCALHOST, port), REQUESTS_IN_PARALLEL, ) .unwrap(); @@ -1111,15 +1249,24 @@ mod tests { let subject = BlockchainInterfaceWeb3::new(transport, event_loop_handle, TEST_DEFAULT_CHAIN); + let start_block = BlockNumber::Number(42u64.into()); let result = subject.retrieve_transactions( - 42, + start_block, + BlockNumber::Latest, &Wallet::from_str("0x3f69f9efd4f2592fd70be8c32ecd9dce71c472fc").unwrap(), ); + let expected_fallback_start_block = + if let BlockNumber::Number(start_block_nbr) = start_block { + start_block_nbr.as_u64() + 1u64 + } else { + panic!("start_block of Latest, Earliest, and Pending are not supported!") + }; + assert_eq!( result, Ok(RetrievedBlockchainTransactions { - new_start_block: 43, + new_start_block: 1 + expected_fallback_start_block, transactions: vec![] }) ); @@ -1134,7 +1281,7 @@ mod tests { ); let (event_loop_handle, transport) = Http::with_max_parallel( - &format!("http://{}:{}", &Ipv4Addr::LOCALHOST.to_string(), port), + &format!("http://{}:{}", &Ipv4Addr::LOCALHOST, port), REQUESTS_IN_PARALLEL, ) .unwrap(); @@ -1157,7 +1304,7 @@ mod tests { { let port = 8545; let (event_loop_handle, transport) = Http::with_max_parallel( - &format!("http://{}:{}", &Ipv4Addr::LOCALHOST.to_string(), port), + &format!("http://{}:{}", &Ipv4Addr::LOCALHOST, port), REQUESTS_IN_PARALLEL, ) .unwrap(); @@ -1180,12 +1327,8 @@ mod tests { vec![br#"{"jsonrpc":"2.0","id":0,"result":"0xFFFQ"}"#.to_vec()], ); - let (event_loop_handle, transport) = Http::new(&format!( - "http://{}:{}", - &Ipv4Addr::LOCALHOST.to_string(), - port - )) - .unwrap(); + let (event_loop_handle, transport) = + Http::new(&format!("http://{}:{}", &Ipv4Addr::LOCALHOST, port)).unwrap(); let subject = BlockchainInterfaceWeb3::new(transport, event_loop_handle, TEST_DEFAULT_CHAIN); @@ -1218,7 +1361,7 @@ mod tests { ]); let (event_loop_handle, transport) = Http::with_max_parallel( - &format!("http://{}:{}", &Ipv4Addr::LOCALHOST.to_string(), port), + &format!("http://{}:{}", &Ipv4Addr::LOCALHOST, port), REQUESTS_IN_PARALLEL, ) .unwrap(); @@ -1240,7 +1383,7 @@ mod tests { ) { let port = 8545; let (event_loop_handle, transport) = Http::with_max_parallel( - &format!("http://{}:{}", &Ipv4Addr::LOCALHOST.to_string(), port), + &format!("http://{}:{}", &Ipv4Addr::LOCALHOST, port), REQUESTS_IN_PARALLEL, ) .unwrap(); @@ -1272,7 +1415,7 @@ mod tests { ]); let (event_loop_handle, transport) = Http::with_max_parallel( - &format!("http://{}:{}", &Ipv4Addr::LOCALHOST.to_string(), port), + &format!("http://{}:{}", &Ipv4Addr::LOCALHOST, port), REQUESTS_IN_PARALLEL, ) .unwrap(); @@ -1293,7 +1436,7 @@ mod tests { "Expected this fragment {} in this err msg: {}", expected_err_msg_fragment, err_msg - ) + ); } #[test] @@ -1745,7 +1888,7 @@ mod tests { assert_gas_limit_is_between(subject, 55000, 65000) } - fn assert_gas_limit_is_between( + fn assert_gas_limit_is_between( mut subject: BlockchainInterfaceWeb3, not_under_this_value: u64, not_above_this_value: u64, @@ -2131,7 +2274,7 @@ mod tests { private_key: H256, } - fn assert_signature(chain: Chain, slice_of_sclices: &[&[u8]]) { + fn assert_signature(chain: Chain, slice_of_slices: &[&[u8]]) { let first_part_tx_1 = r#"[{"nonce": "0x9", "gasPrice": "0x4a817c800", "gasLimit": "0x5208", "to": "0x3535353535353535353535353535353535353535", "value": "0xde0b6b3a7640000", "data": []}, {"private_key": "0x4646464646464646464646464646464646464646464646464646464646464646", "signed": "#; let first_part_tx_2 = r#"[{"nonce": "0x0", "gasPrice": "0xd55698372431", "gasLimit": "0x1e8480", "to": "0xF0109fC8DF283027b6285cc889F5aA624EaC1F55", "value": "0x3b9aca00", "data": []}, {"private_key": "0x4c0883a69102937d6231471b5dbb6204fe5129617082792ae468d01a3f362318", "signed": "#; let first_part_tx_3 = r#"[{"nonce": "0x00", "gasPrice": "0x09184e72a000", "gasLimit": "0x2710", "to": null, "value": "0x00", "data": [127,116,101,115,116,50,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,96,0,87]}, {"private_key": "0xe331b6d69882b4cb4ea581d88e0b604039a3de5967688d3dcffdd2270c0fd109", "signed": "#; @@ -2143,7 +2286,7 @@ mod tests { "[{}]", vec![first_part_tx_1, first_part_tx_2, first_part_tx_3] .iter() - .zip(slice_of_sclices.iter()) + .zip(slice_of_slices.iter()) .zip(0usize..2) .fold(String::new(), |so_far, actual| [ so_far, @@ -2265,7 +2408,7 @@ mod tests { .to_vec() ]); let (event_loop_handle, transport) = Http::with_max_parallel( - &format!("http://{}:{}", &Ipv4Addr::LOCALHOST.to_string(), port), + &format!("http://{}:{}", &Ipv4Addr::LOCALHOST, port), REQUESTS_IN_PARALLEL, ) .unwrap(); @@ -2297,7 +2440,7 @@ mod tests { fn get_transaction_receipt_handles_errors() { let port = find_free_port(); let (event_loop_handle, transport) = Http::with_max_parallel( - &format!("http://{}:{}", &Ipv4Addr::LOCALHOST.to_string(), port), + &format!("http://{}:{}", &Ipv4Addr::LOCALHOST, port), REQUESTS_IN_PARALLEL, ) .unwrap(); @@ -2305,17 +2448,20 @@ mod tests { BlockchainInterfaceWeb3::new(transport, event_loop_handle, TEST_DEFAULT_CHAIN); let tx_hash = make_tx_hash(4564546); - let result = subject.get_transaction_receipt(tx_hash); - - match result { - Err(BlockchainError::QueryFailed(err_message)) => assert!( - err_message.contains("Transport error: Error(Connect, Os"), - "we got this error msg: {}", - err_message - ), - Err(e) => panic!("we expected a different error than: {}", e), - Ok(x) => panic!("we expected an error, but got: {:?}", x), + let actual_error = subject.get_transaction_receipt(tx_hash).unwrap_err(); + let error_message = if let BlockchainError::QueryFailed(em) = actual_error { + em + } else { + panic!("Expected BlockchainError::QueryFailed(msg)"); }; + assert_string_contains( + error_message.as_str(), + "Transport error: Error(Connect, Os { code: ", + ); + assert_string_contains( + error_message.as_str(), + ", kind: ConnectionRefused, message: ", + ); } #[test] @@ -2337,16 +2483,8 @@ mod tests { #[test] fn constant_gwei_matches_calculated_value() { - let value = U256::from(1_000_000_000); - assert_eq!(value.0[0], 1_000_000_000); - assert_eq!(value.0[1], 0); - assert_eq!(value.0[2], 0); - assert_eq!(value.0[3], 0); - let gwei = U256([1_000_000_000u64, 0, 0, 0]); - assert_eq!(value, gwei); assert_eq!(gwei, GWEI); - assert_eq!(value, GWEI); } #[test] @@ -2518,4 +2656,85 @@ mod tests { PayableTransactionError::UninitializedBlockchainInterface ) } + + #[test] + fn non_clandestine_can_fetch_latest_block_number_successfully() { + let prepare_params_arc = Arc::new(Mutex::new(vec![])); + let transport = TestTransport::default() + .prepare_params(&prepare_params_arc) + .send_result(json!("0x1e37066")); + let subject = BlockchainInterfaceWeb3::new( + transport, + make_fake_event_loop_handle(), + TEST_DEFAULT_CHAIN, + ); + + let latest_block_number = subject.get_block_number().unwrap(); + + assert_eq!(latest_block_number, U64::from(0x1e37066u64)); + + let mut prepare_params = prepare_params_arc.lock().unwrap(); + let (method_name, actual_arguments) = prepare_params.remove(0); + assert!(prepare_params.is_empty()); + assert_eq!(method_name, "eth_blockNumber".to_string()); + let expected_arguments: Vec = vec![]; + assert_eq!(actual_arguments, expected_arguments); + } + + #[test] + fn non_clandestine_can_handle_latest_null_block_number_error() { + let prepare_params_arc = Arc::new(Mutex::new(vec![])); + let transport = TestTransport::default() + .prepare_params(&prepare_params_arc) + .send_result(Value::Null); + let subject = BlockchainInterfaceWeb3::new( + transport, + make_fake_event_loop_handle(), + TEST_DEFAULT_CHAIN, + ); + + let expected_error = subject.get_block_number().unwrap_err(); + assert_eq!( + expected_error, + BlockchainError::QueryFailed("Decoder error: Error(\"invalid type: null, expected a 0x-prefixed hex string with length between (0; 16]\", line: 0, column: 0)".to_string()) + ); + + let mut prepare_params = prepare_params_arc.lock().unwrap(); + let (method_name, actual_arguments) = prepare_params.remove(0); + assert!(prepare_params.is_empty()); + assert_eq!(method_name, "eth_blockNumber".to_string()); + let expected_arguments: Vec = vec![]; + assert_eq!(actual_arguments, expected_arguments); + } + + #[test] + fn non_clandestine_can_handle_latest_string_block_number_error() { + let prepare_params_arc: Arc)>>> = + Arc::new(Mutex::new(vec![])); + let transport = TestTransport::default() + .prepare_params(&prepare_params_arc) + .send_result(Value::String("this is an invalid block number".to_string())); + + let subject = BlockchainInterfaceWeb3::new( + transport.clone(), + make_fake_event_loop_handle(), + TEST_DEFAULT_CHAIN, + ); + + let expected_error = subject.get_block_number().unwrap_err(); + + assert_eq!( + expected_error, + BlockchainError::QueryFailed( + "Decoder error: Error(\"0x prefix is missing\", line: 0, column: 0)".to_string() + ) + ); + + let mut prepare_params = prepare_params_arc.lock().unwrap(); + let (method_name, actual_arguments) = prepare_params.remove(0); + assert!(prepare_params.is_empty()); + assert_eq!(method_name, "eth_blockNumber".to_string()); + let expected_arguments: Vec = vec![]; + assert_eq!(actual_arguments, expected_arguments); + } } diff --git a/node/src/blockchain/test_utils.rs b/node/src/blockchain/test_utils.rs index 007a5b51b..9778cb686 100644 --- a/node/src/blockchain/test_utils.rs +++ b/node/src/blockchain/test_utils.rs @@ -4,9 +4,9 @@ use crate::blockchain::blockchain_bridge::PendingPayableFingerprintSeeds; use crate::blockchain::blockchain_interface::{ - BlockchainError, BlockchainInterface, BlockchainResult, PayableTransactionError, - ProcessedPayableFallible, ResultForBalance, ResultForNonce, ResultForReceipt, - REQUESTS_IN_PARALLEL, + BlockchainError, BlockchainInterface, BlockchainResult, LatestBlockNumber, + PayableTransactionError, ProcessedPayableFallible, ResultForBalance, ResultForNonce, + ResultForReceipt, REQUESTS_IN_PARALLEL, }; use crate::sub_lib::wallet::Wallet; use actix::Recipient; @@ -23,7 +23,7 @@ use std::time::SystemTime; use crate::accountant::db_access_objects::payable_dao::PayableAccount; use crate::blockchain::batch_payable_tools::BatchPayableTools; use web3::transports::{Batch, EventLoopHandle, Http}; -use web3::types::{Address, Bytes, SignedTransaction, TransactionParameters, U256}; +use web3::types::{Address, BlockNumber, Bytes, SignedTransaction, TransactionParameters, U256}; use web3::{BatchTransport, Error as Web3Error, Web3}; use web3::{RequestId, Transport}; @@ -55,7 +55,7 @@ pub fn make_meaningless_seed() -> Seed { #[derive(Default)] pub struct BlockchainInterfaceMock { - retrieve_transactions_parameters: Arc>>, + retrieve_transactions_parameters: Arc>>, retrieve_transactions_results: RefCell>>, send_payables_within_batch_params: Arc< @@ -80,6 +80,7 @@ pub struct BlockchainInterfaceMock { contract_address_results: RefCell>, get_transaction_count_parameters: Arc>>, get_transaction_count_results: RefCell>>, + get_block_number_results: RefCell>, } impl BlockchainInterface for BlockchainInterfaceMock { @@ -89,13 +90,15 @@ impl BlockchainInterface for BlockchainInterfaceMock { fn retrieve_transactions( &self, - start_block: u64, + start_block: BlockNumber, + end_block: BlockNumber, recipient: &Wallet, ) -> Result { - self.retrieve_transactions_parameters - .lock() - .unwrap() - .push((start_block, recipient.clone())); + self.retrieve_transactions_parameters.lock().unwrap().push(( + start_block, + end_block, + recipient.clone(), + )); self.retrieve_transactions_results.borrow_mut().remove(0) } @@ -122,6 +125,10 @@ impl BlockchainInterface for BlockchainInterfaceMock { .remove(0) } + fn get_block_number(&self) -> LatestBlockNumber { + self.get_block_number_results.borrow_mut().remove(0) + } + fn get_transaction_fee_balance(&self, address: &Wallet) -> ResultForBalance { self.get_transaction_fee_balance_params .lock() @@ -158,7 +165,10 @@ impl BlockchainInterface for BlockchainInterfaceMock { } impl BlockchainInterfaceMock { - pub fn retrieve_transactions_params(mut self, params: &Arc>>) -> Self { + pub fn retrieve_transactions_params( + mut self, + params: &Arc>>, + ) -> Self { self.retrieve_transactions_parameters = params.clone(); self } @@ -199,6 +209,11 @@ impl BlockchainInterfaceMock { self } + pub fn get_block_number_result(self, result: LatestBlockNumber) -> Self { + self.get_block_number_results.borrow_mut().push(result); + self + } + pub fn get_transaction_fee_balance_params(mut self, params: &Arc>>) -> Self { self.get_transaction_fee_balance_params = params.clone(); self diff --git a/node/src/database/db_initializer.rs b/node/src/database/db_initializer.rs index e961a631c..2fad52508 100644 --- a/node/src/database/db_initializer.rs +++ b/node/src/database/db_initializer.rs @@ -261,6 +261,7 @@ impl DbInitializerReal { false, "scan intervals", ); + Self::set_config_value(conn, "max_block_count", None, false, "maximum block count"); } fn create_pending_payable_table(&self, conn: &Connection) { @@ -764,7 +765,7 @@ mod tests { #[test] fn constants_have_correct_values() { assert_eq!(DATABASE_FILE, "node-data.db"); - assert_eq!(CURRENT_SCHEMA_VERSION, 8); + assert_eq!(CURRENT_SCHEMA_VERSION, 9); } #[test] @@ -1039,6 +1040,7 @@ mod tests { false, ); verify(&mut config_vec, "mapping_protocol", None, false); + verify(&mut config_vec, "max_block_count", None, false); verify(&mut config_vec, "min_hops", Some("3"), false); verify( &mut config_vec, diff --git a/node/src/database/db_migrations/db_migrator.rs b/node/src/database/db_migrations/db_migrator.rs index 0866a7d0d..4dba9ce97 100644 --- a/node/src/database/db_migrations/db_migrator.rs +++ b/node/src/database/db_migrations/db_migrator.rs @@ -10,6 +10,7 @@ use crate::database::db_migrations::migrations::migration_4_to_5::Migrate_4_to_5 use crate::database::db_migrations::migrations::migration_5_to_6::Migrate_5_to_6; use crate::database::db_migrations::migrations::migration_6_to_7::Migrate_6_to_7; use crate::database::db_migrations::migrations::migration_7_to_8::Migrate_7_to_8; +use crate::database::db_migrations::migrations::migration_8_to_9::Migrate_8_to_9; use crate::database::db_migrations::migrator_utils::{ DBMigDeclarator, DBMigrationUtilities, DBMigrationUtilitiesReal, DBMigratorInnerConfiguration, }; @@ -77,6 +78,7 @@ impl DbMigratorReal { &Migrate_5_to_6, &Migrate_6_to_7, &Migrate_7_to_8, + &Migrate_8_to_9, ] } diff --git a/node/src/database/db_migrations/migrations/migration_8_to_9.rs b/node/src/database/db_migrations/migrations/migration_8_to_9.rs new file mode 100644 index 000000000..c5928edb6 --- /dev/null +++ b/node/src/database/db_migrations/migrations/migration_8_to_9.rs @@ -0,0 +1,70 @@ +use crate::database::db_migrations::db_migrator::DatabaseMigration; +use crate::database::db_migrations::migrator_utils::DBMigDeclarator; + +#[allow(non_camel_case_types)] +pub struct Migrate_8_to_9; + +impl DatabaseMigration for Migrate_8_to_9 { + fn migrate<'a>( + &self, + declaration_utils: Box, + ) -> rusqlite::Result<()> { + declaration_utils.execute_upon_transaction(&[ + &"INSERT INTO config (name, value, encrypted) VALUES ('max_block_count', null, 0)", + ]) + } + + fn old_version(&self) -> usize { + 8 + } +} + +#[cfg(test)] +mod tests { + use crate::database::db_initializer::{ + DbInitializationConfig, DbInitializer, DbInitializerReal, DATABASE_FILE, + }; + use crate::test_utils::database_utils::{ + bring_db_0_back_to_life_and_return_connection, make_external_data, retrieve_config_row, + }; + use masq_lib::test_utils::logging::{init_test_logging, TestLogHandler}; + use masq_lib::test_utils::utils::ensure_node_home_directory_exists; + use std::fs::create_dir_all; + + #[test] + fn migration_from_8_to_9_is_properly_set() { + init_test_logging(); + let dir_path = ensure_node_home_directory_exists( + "db_migrations", + "migration_from_8_to_9_is_properly_set", + ); + create_dir_all(&dir_path).unwrap(); + let db_path = dir_path.join(DATABASE_FILE); + let _ = bring_db_0_back_to_life_and_return_connection(&db_path); + let subject = DbInitializerReal::default(); + + let result = subject.initialize_to_version( + &dir_path, + 9, + DbInitializationConfig::create_or_migrate(make_external_data()), + ); + let connection = result.unwrap(); + let (mp_value, mp_encrypted) = retrieve_config_row(connection.as_ref(), "max_block_count"); + let (cs_value, cs_encrypted) = retrieve_config_row(connection.as_ref(), "schema_version"); + assert_eq!(mp_value, None); + assert_eq!(mp_encrypted, false); + assert_eq!(cs_value, Some("9".to_string())); + assert_eq!(cs_encrypted, false); + TestLogHandler::new().assert_logs_contain_in_order(vec![ + "DbMigrator: Database successfully migrated from version 0 to 1", + "DbMigrator: Database successfully migrated from version 1 to 2", + "DbMigrator: Database successfully migrated from version 2 to 3", + "DbMigrator: Database successfully migrated from version 3 to 4", + "DbMigrator: Database successfully migrated from version 4 to 5", + "DbMigrator: Database successfully migrated from version 5 to 6", + "DbMigrator: Database successfully migrated from version 6 to 7", + "DbMigrator: Database successfully migrated from version 7 to 8", + "DbMigrator: Database successfully migrated from version 8 to 9", + ]); + } +} diff --git a/node/src/database/db_migrations/migrations/mod.rs b/node/src/database/db_migrations/migrations/mod.rs index bc540b4ae..68b10ca9b 100644 --- a/node/src/database/db_migrations/migrations/mod.rs +++ b/node/src/database/db_migrations/migrations/mod.rs @@ -8,3 +8,4 @@ pub mod migration_4_to_5; pub mod migration_5_to_6; pub mod migration_6_to_7; pub mod migration_7_to_8; +pub mod migration_8_to_9; diff --git a/node/src/db_config/config_dao.rs b/node/src/db_config/config_dao.rs index 6bd5aef9c..2e96d350d 100644 --- a/node/src/db_config/config_dao.rs +++ b/node/src/db_config/config_dao.rs @@ -266,4 +266,16 @@ mod tests { let result = subject.get("schema_version").unwrap(); assert_eq!(result, ConfigDaoRecord::new("schema_version", None, false)); } + + #[test] + fn test_handle_update_execution() { + let result = handle_update_execution(Err(rusqlite::Error::ExecuteReturnedResults)); + + assert_eq!( + result, + Err(ConfigDaoError::DatabaseError( + "Execute returned results - did you mean to call query?".to_string() + )) + ) + } } diff --git a/node/src/db_config/config_dao_null.rs b/node/src/db_config/config_dao_null.rs index e3550da14..2cc00e527 100644 --- a/node/src/db_config/config_dao_null.rs +++ b/node/src/db_config/config_dao_null.rs @@ -130,6 +130,7 @@ impl Default for ConfigDaoNull { "scan_intervals".to_string(), (Some(DEFAULT_SCAN_INTERVALS.to_string()), false), ); + data.insert("max_block_count".to_string(), (None, false)); Self { data } } } @@ -273,6 +274,7 @@ mod tests { "schema_version", Some(format!("{}", CURRENT_SCHEMA_VERSION).as_str()), ), + ("max_block_count", None), ] .into_iter() .map(|(k, v_opt)| (k.to_string(), v_opt.map(|v| v.to_string()))) diff --git a/node/src/db_config/persistent_configuration.rs b/node/src/db_config/persistent_configuration.rs index 6cdef5e7c..b16e280c4 100644 --- a/node/src/db_config/persistent_configuration.rs +++ b/node/src/db_config/persistent_configuration.rs @@ -136,6 +136,8 @@ pub trait PersistentConfiguration { ) -> Result<(), PersistentConfigError>; fn start_block(&self) -> Result; fn set_start_block(&mut self, value: u64) -> Result<(), PersistentConfigError>; + fn max_block_count(&self) -> Result, PersistentConfigError>; + fn set_max_block_count(&mut self, value: Option) -> Result<(), PersistentConfigError>; fn set_wallet_info( &mut self, consuming_wallet_private_key: &str, @@ -413,6 +415,14 @@ impl PersistentConfiguration for PersistentConfigurationReal { self.simple_set_method("start_block", value) } + fn max_block_count(&self) -> Result, PersistentConfigError> { + Ok(decode_u64(self.get("max_block_count")?)?) + } + + fn set_max_block_count(&mut self, value: Option) -> Result<(), PersistentConfigError> { + Ok(self.dao.set("max_block_count", encode_u64(value)?)?) + } + fn set_wallet_info( &mut self, consuming_wallet_private_key: &str, @@ -1935,6 +1945,39 @@ mod tests { ); } + #[test] + fn max_block_count_set_method_works_with_some() { + let set_params_arc = Arc::new(Mutex::new(Vec::new())); + let config_dao = ConfigDaoMock::new() + .set_params(&set_params_arc) + .set_result(Ok(())); + let mut subject = PersistentConfigurationReal::new(Box::new(config_dao)); + + let result = subject.set_max_block_count(Some(100_000u64)); + + assert!(result.is_ok()); + let set_params = set_params_arc.lock().unwrap(); + assert_eq!( + *set_params, + vec![("max_block_count".to_string(), Some(100_000u64.to_string()))] + ); + } + + #[test] + fn max_block_count_set_method_works_with_none() { + let set_params_arc = Arc::new(Mutex::new(Vec::new())); + let config_dao = ConfigDaoMock::new() + .set_params(&set_params_arc) + .set_result(Ok(())); + let mut subject = PersistentConfigurationReal::new(Box::new(config_dao)); + + let result = subject.set_max_block_count(None); + + assert!(result.is_ok()); + let set_params = set_params_arc.lock().unwrap(); + assert_eq!(*set_params, vec![("max_block_count".to_string(), None)]); + } + #[test] #[should_panic( expected = "ever-supplied value missing: payment_thresholds; database is corrupt!" diff --git a/node/src/node_configurator/configurator.rs b/node/src/node_configurator/configurator.rs index ab8329e2e..002bfd27d 100644 --- a/node/src/node_configurator/configurator.rs +++ b/node/src/node_configurator/configurator.rs @@ -554,6 +554,13 @@ impl Configurator { "earningWalletAddressOpt", )?; let start_block = Self::value_required(persistent_config.start_block(), "startBlock")?; + let max_block_count_opt = match persistent_config.max_block_count() { + Ok(value) => value, + Err(e) => panic!( + "Database corruption: Could not read max block count: {:?}", + e + ), + }; let neighborhood_mode = Self::value_required(persistent_config.neighborhood_mode(), "neighborhoodMode")? .to_string(); @@ -623,6 +630,7 @@ impl Configurator { clandestine_port, chain_name, gas_price, + max_block_count_opt, neighborhood_mode, consuming_wallet_private_key_opt, consuming_wallet_address_opt, @@ -2417,6 +2425,7 @@ mod tests { .gas_price_result(Ok(2345)) .consuming_wallet_private_key_result(Ok(Some(consuming_wallet_private_key))) .mapping_protocol_result(Ok(Some(AutomapProtocol::Igdp))) + .max_block_count_result(Ok(Some(100000))) .neighborhood_mode_result(Ok(NeighborhoodModeLight::Standard)) .past_neighbors_result(Ok(Some(vec![node_descriptor.clone()]))) .earning_wallet_address_result(Ok(Some(earning_wallet_address.clone()))) @@ -2442,6 +2451,7 @@ mod tests { clandestine_port: 1234, chain_name: "ropsten".to_string(), gas_price: 2345, + max_block_count_opt: Some(100000), neighborhood_mode: String::from("standard"), consuming_wallet_private_key_opt: None, consuming_wallet_address_opt: None, @@ -2545,6 +2555,7 @@ mod tests { .consuming_wallet_private_key_params(&consuming_wallet_private_key_params_arc) .consuming_wallet_private_key_result(Ok(Some(consuming_wallet_private_key.clone()))) .mapping_protocol_result(Ok(Some(AutomapProtocol::Igdp))) + .max_block_count_result(Ok(None)) .neighborhood_mode_result(Ok(NeighborhoodModeLight::ConsumeOnly)) .past_neighbors_params(&past_neighbors_params_arc) .past_neighbors_result(Ok(Some(vec![node_descriptor.clone()]))) @@ -2572,6 +2583,7 @@ mod tests { clandestine_port: 1234, chain_name: "ropsten".to_string(), gas_price: 2345, + max_block_count_opt: None, neighborhood_mode: String::from("consume-only"), consuming_wallet_private_key_opt: Some(consuming_wallet_private_key), consuming_wallet_address_opt: Some(consuming_wallet_address), @@ -2610,6 +2622,118 @@ mod tests { assert_eq!(*past_neighbors_params, vec!["password".to_string()]) } + #[test] + fn configuration_handles_retrieving_all_possible_none_values() { + let persistent_config = PersistentConfigurationMock::new() + .blockchain_service_url_result(Ok(None)) + .current_schema_version_result("3") + .clandestine_port_result(Ok(1234)) + .chain_name_result("ropsten".to_string()) + .gas_price_result(Ok(2345)) + .earning_wallet_address_result(Ok(None)) + .start_block_result(Ok(3456)) + .max_block_count_result(Ok(None)) + .neighborhood_mode_result(Ok(NeighborhoodModeLight::ZeroHop)) + .mapping_protocol_result(Ok(None)) + .consuming_wallet_private_key_result(Ok(None)) + .past_neighbors_result(Ok(None)) + .rate_pack_result(Ok(RatePack { + routing_byte_rate: 0, + routing_service_rate: 0, + exit_byte_rate: 0, + exit_service_rate: 0, + })) + .scan_intervals_result(Ok(ScanIntervals { + pending_payable_scan_interval: Default::default(), + payable_scan_interval: Default::default(), + receivable_scan_interval: Default::default(), + })) + .payment_thresholds_result(Ok(PaymentThresholds { + debt_threshold_gwei: 0, + maturity_threshold_sec: 0, + payment_grace_period_sec: 0, + permanent_debt_allowed_gwei: 0, + threshold_interval_sec: 0, + unban_below_gwei: 0, + })); + let mut subject = make_subject(Some(persistent_config)); + + let (configuration, context_id) = + UiConfigurationResponse::fmb(subject.handle_configuration( + UiConfigurationRequest { + db_password_opt: None, + }, + 4321, + )) + .unwrap(); + + assert_eq!(context_id, 4321); + assert_eq!( + configuration, + UiConfigurationResponse { + blockchain_service_url_opt: None, + current_schema_version: "3".to_string(), + clandestine_port: 1234, + chain_name: "ropsten".to_string(), + gas_price: 2345, + max_block_count_opt: None, + neighborhood_mode: String::from("zero-hop"), + consuming_wallet_private_key_opt: None, + consuming_wallet_address_opt: None, + earning_wallet_address_opt: None, + port_mapping_protocol_opt: None, + past_neighbors: vec![], + payment_thresholds: UiPaymentThresholds { + threshold_interval_sec: 0, + debt_threshold_gwei: 0, + maturity_threshold_sec: 0, + payment_grace_period_sec: 0, + permanent_debt_allowed_gwei: 0, + unban_below_gwei: 0 + }, + rate_pack: UiRatePack { + routing_byte_rate: 0, + routing_service_rate: 0, + exit_byte_rate: 0, + exit_service_rate: 0 + }, + start_block: 3456, + scan_intervals: UiScanIntervals { + pending_payable_sec: 0, + payable_sec: 0, + receivable_sec: 0 + } + } + ); + } + + #[test] + #[should_panic( + expected = "Database corruption: Could not read max block count: DatabaseError(\"Corruption\")" + )] + fn configuration_panic_on_error_retrieving_max_block_count() { + let persistent_config = PersistentConfigurationMock::new() + .check_password_result(Ok(true)) + .blockchain_service_url_result(Ok(None)) + .current_schema_version_result("3") + .clandestine_port_result(Ok(1234)) + .chain_name_result("ropsten".to_string()) + .gas_price_result(Ok(2345)) + .earning_wallet_address_result(Ok(Some("4a5e43b54c6C56Ebf7".to_string()))) + .start_block_result(Ok(3456)) + .max_block_count_result(Err(PersistentConfigError::DatabaseError( + "Corruption".to_string(), + ))); + let mut subject = make_subject(Some(persistent_config)); + + let _result = subject.handle_configuration( + UiConfigurationRequest { + db_password_opt: Some("password".to_string()), + }, + 4321, + ); + } + #[test] fn configuration_handles_check_password_error() { let persistent_config = PersistentConfigurationMock::new() @@ -2647,6 +2771,7 @@ mod tests { "0x0123456789012345678901234567890123456789".to_string(), ))) .start_block_result(Ok(3456)) + .max_block_count_result(Ok(Some(100000))) .neighborhood_mode_result(Ok(NeighborhoodModeLight::ConsumeOnly)) .mapping_protocol_result(Ok(Some(AutomapProtocol::Igdp))) .consuming_wallet_private_key_result(cwpk); diff --git a/node/src/proxy_server/mod.rs b/node/src/proxy_server/mod.rs index 4149a09e0..209548f6d 100644 --- a/node/src/proxy_server/mod.rs +++ b/node/src/proxy_server/mod.rs @@ -236,10 +236,11 @@ impl ProxyServer { } fn handle_dns_resolve_failure(&mut self, msg: &ExpiredCoresPackage) { - let return_route_info = match self.get_return_route_info(&msg.remaining_route) { - Some(rri) => rri, - None => return, // TODO: Eventually we'll have to do something better here, but we'll probably need some heuristics. - }; + let return_route_info = + match self.get_return_route_info(&msg.remaining_route, "dns resolve failure") { + Some(rri) => rri, + None => return, // TODO: Eventually we'll have to do something better here, but we'll probably need some heuristics. + }; let exit_public_key = { // ugly, ugly let self_public_key = self.main_cryptde.public_key(); @@ -325,10 +326,11 @@ impl ProxyServer { "Relaying ClientResponsePayload (stream key {}, sequence {}, length {}) from Hopper to Dispatcher for client", response.stream_key, response.sequenced_packet.sequence_number, response.sequenced_packet.data.len() ); - let return_route_info = match self.get_return_route_info(&msg.remaining_route) { - Some(rri) => rri, - None => return, - }; + let return_route_info = + match self.get_return_route_info(&msg.remaining_route, "client response") { + Some(rri) => rri, + None => return, + }; self.report_response_services_consumed( &return_route_info, response.sequenced_packet.data.len(), @@ -738,7 +740,11 @@ impl ProxyServer { } } - fn get_return_route_info(&self, remaining_route: &Route) -> Option> { + fn get_return_route_info( + &self, + remaining_route: &Route, + source: &str, + ) -> Option> { let mut mut_remaining_route = remaining_route.clone(); mut_remaining_route .shift(self.main_cryptde) @@ -753,7 +759,7 @@ impl ProxyServer { match self.route_ids_to_return_routes.get(&return_route_id) { Some(rri) => Some(rri), None => { - error!(self.logger, "Can't report services consumed: received response with bogus return-route ID {}. Ignoring", return_route_id); + error!(self.logger, "Can't report services consumed: received response with bogus return-route ID {} for {}. Ignoring", return_route_id, source); None } } @@ -4188,7 +4194,7 @@ mod tests { System::current().stop(); system.run(); - TestLogHandler::new().exists_log_containing("ERROR: ProxyServer: Can't report services consumed: received response with bogus return-route ID 1234. Ignoring"); + TestLogHandler::new().exists_log_containing("ERROR: ProxyServer: Can't report services consumed: received response with bogus return-route ID 1234 for client response. Ignoring"); assert_eq!(dispatcher_recording_arc.lock().unwrap().len(), 0); assert_eq!(accountant_recording_arc.lock().unwrap().len(), 0); } @@ -4307,7 +4313,7 @@ mod tests { ); subject_addr.try_send(expired_cores_package).unwrap(); - TestLogHandler::new().await_log_containing("ERROR: ProxyServer: Can't report services consumed: received response with bogus return-route ID 1234. Ignoring", 1000); + TestLogHandler::new().await_log_containing("ERROR: ProxyServer: Can't report services consumed: received response with bogus return-route ID 1234 for client response. Ignoring", 1000); } #[test] diff --git a/node/src/test_utils/persistent_configuration_mock.rs b/node/src/test_utils/persistent_configuration_mock.rs index 1f8fceadc..e93ce308c 100644 --- a/node/src/test_utils/persistent_configuration_mock.rs +++ b/node/src/test_utils/persistent_configuration_mock.rs @@ -12,6 +12,7 @@ use masq_lib::utils::AutomapProtocol; use masq_lib::utils::NeighborhoodModeLight; use std::cell::RefCell; use std::sync::{Arc, Mutex}; +use std::u64; #[allow(clippy::type_complexity)] #[derive(Clone, Default)] @@ -59,6 +60,10 @@ pub struct PersistentConfigurationMock { start_block_results: RefCell>>, set_start_block_params: Arc>>, set_start_block_results: RefCell>>, + max_block_count_params: Arc>>, + max_block_count_results: RefCell, PersistentConfigError>>>, + set_max_block_count_params: Arc>>>, + set_max_block_count_results: RefCell>>, payment_thresholds_results: RefCell>>, set_payment_thresholds_params: Arc>>, set_payment_thresholds_results: RefCell>>, @@ -232,6 +237,16 @@ impl PersistentConfiguration for PersistentConfigurationMock { Self::result_from(&self.set_start_block_results) } + fn max_block_count(&self) -> Result, PersistentConfigError> { + self.max_block_count_params.lock().unwrap().push(()); + Self::result_from(&self.max_block_count_results) + } + + fn set_max_block_count(&mut self, value: Option) -> Result<(), PersistentConfigError> { + self.set_max_block_count_params.lock().unwrap().push(value); + Self::result_from(&self.set_max_block_count_results) + } + fn set_wallet_info( &mut self, consuming_wallet_private_key: &str, @@ -312,7 +327,7 @@ impl PersistentConfigurationMock { self } - pub fn current_schema_version_result(self, result: &str) -> PersistentConfigurationMock { + pub fn current_schema_version_result(self, result: &str) -> Self { self.current_schema_version_results .borrow_mut() .push(result.to_string()); @@ -333,55 +348,37 @@ impl PersistentConfigurationMock { pub fn change_password_params( mut self, params: &Arc, String)>>>, - ) -> PersistentConfigurationMock { + ) -> Self { self.change_password_params = params.clone(); self } - pub fn change_password_result( - self, - result: Result<(), PersistentConfigError>, - ) -> PersistentConfigurationMock { + pub fn change_password_result(self, result: Result<(), PersistentConfigError>) -> Self { self.change_password_results.borrow_mut().push(result); self } - pub fn check_password_params( - mut self, - params: &Arc>>>, - ) -> PersistentConfigurationMock { + pub fn check_password_params(mut self, params: &Arc>>>) -> Self { self.check_password_params = params.clone(); self } - pub fn check_password_result( - self, - result: Result, - ) -> PersistentConfigurationMock { + pub fn check_password_result(self, result: Result) -> Self { self.check_password_results.borrow_mut().push(result); self } - pub fn clandestine_port_result( - self, - result: Result, - ) -> PersistentConfigurationMock { + pub fn clandestine_port_result(self, result: Result) -> Self { self.clandestine_port_results.borrow_mut().push(result); self } - pub fn set_clandestine_port_params( - mut self, - params: &Arc>>, - ) -> PersistentConfigurationMock { + pub fn set_clandestine_port_params(mut self, params: &Arc>>) -> Self { self.set_clandestine_port_params = params.clone(); self } - pub fn set_clandestine_port_result( - self, - result: Result<(), PersistentConfigError>, - ) -> PersistentConfigurationMock { + pub fn set_clandestine_port_result(self, result: Result<(), PersistentConfigError>) -> Self { self.set_clandestine_port_results.borrow_mut().push(result); self } @@ -410,7 +407,7 @@ impl PersistentConfigurationMock { pub fn neighborhood_mode_result( self, result: Result, - ) -> PersistentConfigurationMock { + ) -> Self { self.neighborhood_mode_results.borrow_mut().push(result); self } @@ -418,23 +415,17 @@ impl PersistentConfigurationMock { pub fn set_neighborhood_mode_params( mut self, params: &Arc>>, - ) -> PersistentConfigurationMock { + ) -> Self { self.set_neighborhood_mode_params = params.clone(); self } - pub fn set_neighborhood_mode_result( - self, - result: Result<(), PersistentConfigError>, - ) -> PersistentConfigurationMock { + pub fn set_neighborhood_mode_result(self, result: Result<(), PersistentConfigError>) -> Self { self.set_neighborhood_mode_results.borrow_mut().push(result); self } - pub fn consuming_wallet_params( - mut self, - params: &Arc>>, - ) -> PersistentConfigurationMock { + pub fn consuming_wallet_params(mut self, params: &Arc>>) -> Self { self.consuming_wallet_params = params.clone(); self } @@ -442,15 +433,12 @@ impl PersistentConfigurationMock { pub fn consuming_wallet_result( self, result: Result, PersistentConfigError>, - ) -> PersistentConfigurationMock { + ) -> Self { self.consuming_wallet_results.borrow_mut().push(result); self } - pub fn consuming_wallet_private_key_params( - mut self, - params: &Arc>>, - ) -> PersistentConfigurationMock { + pub fn consuming_wallet_private_key_params(mut self, params: &Arc>>) -> Self { self.consuming_wallet_private_key_params = params.clone(); self } @@ -458,7 +446,7 @@ impl PersistentConfigurationMock { pub fn consuming_wallet_private_key_result( self, result: Result, PersistentConfigError>, - ) -> PersistentConfigurationMock { + ) -> Self { self.consuming_wallet_private_key_results .borrow_mut() .push(result); @@ -469,7 +457,7 @@ impl PersistentConfigurationMock { pub fn set_wallet_info_params( mut self, params: &Arc>>, - ) -> PersistentConfigurationMock { + ) -> Self { self.set_wallet_info_params = params.clone(); self } @@ -484,10 +472,7 @@ impl PersistentConfigurationMock { self } - pub fn set_gas_price_params( - mut self, - params: &Arc>>, - ) -> PersistentConfigurationMock { + pub fn set_gas_price_params(mut self, params: &Arc>>) -> Self { self.set_gas_price_params = params.clone(); self } @@ -497,10 +482,7 @@ impl PersistentConfigurationMock { self } - pub fn past_neighbors_params( - mut self, - params: &Arc>>, - ) -> PersistentConfigurationMock { + pub fn past_neighbors_params(mut self, params: &Arc>>) -> Self { self.past_neighbors_params = params.clone(); self } @@ -508,7 +490,7 @@ impl PersistentConfigurationMock { pub fn past_neighbors_result( self, result: Result>, PersistentConfigError>, - ) -> PersistentConfigurationMock { + ) -> Self { self.past_neighbors_results.borrow_mut().push(result); self } @@ -517,15 +499,12 @@ impl PersistentConfigurationMock { pub fn set_past_neighbors_params( mut self, params: &Arc>, String)>>>, - ) -> PersistentConfigurationMock { + ) -> Self { self.set_past_neighbors_params = params.clone(); self } - pub fn set_past_neighbors_result( - self, - result: Result<(), PersistentConfigError>, - ) -> PersistentConfigurationMock { + pub fn set_past_neighbors_result(self, result: Result<(), PersistentConfigError>) -> Self { self.set_past_neighbors_results.borrow_mut().push(result); self } @@ -533,7 +512,7 @@ impl PersistentConfigurationMock { pub fn earning_wallet_result( self, result: Result, PersistentConfigError>, - ) -> PersistentConfigurationMock { + ) -> Self { self.earning_wallet_results.borrow_mut().push(result); self } @@ -541,7 +520,7 @@ impl PersistentConfigurationMock { pub fn earning_wallet_address_result( self, result: Result, PersistentConfigError>, - ) -> PersistentConfigurationMock { + ) -> Self { self.earning_wallet_address_results .borrow_mut() .push(result); @@ -558,10 +537,7 @@ impl PersistentConfigurationMock { self } - pub fn set_start_block_params( - mut self, - params: &Arc>>, - ) -> PersistentConfigurationMock { + pub fn set_start_block_params(mut self, params: &Arc>>) -> Self { self.set_start_block_params = params.clone(); self } @@ -571,6 +547,29 @@ impl PersistentConfigurationMock { self } + pub fn max_block_count_params(mut self, params: &Arc>>) -> Self { + self.max_block_count_params = params.clone(); + self + } + + pub fn max_block_count_result( + self, + result: Result, PersistentConfigError>, + ) -> Self { + self.max_block_count_results.borrow_mut().push(result); + self + } + + pub fn set_max_block_count_params(mut self, params: &Arc>>>) -> Self { + self.set_max_block_count_params = params.clone(); + self + } + + pub fn set_max_block_count_result(self, result: Result<(), PersistentConfigError>) -> Self { + self.set_max_block_count_results.borrow_mut().push(result); + self + } + pub fn payment_thresholds_result( self, result: Result,