diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 488c696a1..c73cd41f5 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -15,7 +15,7 @@ jobs: uses: sudo-bot/action-docker-compose@latest with: # https://docs.docker.com/compose/reference/overview/ - cli-args: "-f docker-compose.ci.yml up -d" + cli-args: "-f docker-compose.harness.yml up -d --build" - uses: actions-rs/toolchain@v1 with: # No need to add `toolchain`, it will use `rust-toolchain` file instead @@ -26,10 +26,4 @@ jobs: - name: Run `cargo make ci-flow` # Running cargo make doesn't successfully start `ganache` run: | - cargo make ci-flow - # set environment variables for `primitives` postgres tests - env: - POSTGRES_USER: postgres - POSTGRES_PASSWORD: postgres - POSTGRES_HOST: localhost - POSTGRES_DB: sentry_leader + cargo make ci-flow \ No newline at end of file diff --git a/Cargo.lock b/Cargo.lock index 00867c83b..baad23faa 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7,7 +7,7 @@ name = "adapter" version = "0.1.0" dependencies = [ "async-trait", - "base64 0.13.0", + "base64", "byteorder 1.4.3", "chrono", "create2", @@ -16,7 +16,7 @@ dependencies = [ "ethstore", "futures", "hex", - "lazy_static", + "once_cell", "primitives", "reqwest", "serde", @@ -349,12 +349,6 @@ version = "0.2.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a4521f3e3d031370679b3b140beb36dfe4801b09ac77e30c61941f97df3ef28b" -[[package]] -name = "base64" -version = "0.12.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3441f0f7b02788e948e47f457ca01f1d7e6d92c693bc132c22b087d3141c03ff" - [[package]] name = "base64" version = "0.13.0" @@ -532,12 +526,6 @@ version = "1.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "14c189c53d098945499cdfa7ecc63567cf3886b3332b312a5b4585d8d3a6a610" -[[package]] -name = "bytes" -version = "0.5.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0e4cec68f03f32e44924783795810fa50a7035d8c8ebe78580ad7e6c703fba38" - [[package]] name = "bytes" version = "1.0.1" @@ -623,7 +611,7 @@ version = "4.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a2d47c1b11006b87e492b53b313bb699ce60e16613c4dddaa91f8f7c220ab2fa" dependencies = [ - "bytes 1.0.1", + "bytes", "futures-util", "memchr", "pin-project-lite", @@ -1022,6 +1010,15 @@ dependencies = [ "cfg-if 1.0.0", ] +[[package]] +name = "envy" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f47e0157f2cb54f5ae1bd371b30a2ae4311e1c028f575cd4e81de7353215965" +dependencies = [ + "serde", +] + [[package]] name = "error-chain" version = "0.12.4" @@ -1473,7 +1470,7 @@ version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d832b01df74254fe364568d6ddc294443f61cbec82816b60904303af87efae78" dependencies = [ - "bytes 1.0.1", + "bytes", "fnv", "futures-core", "futures-sink", @@ -1498,9 +1495,9 @@ version = "0.3.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0b7591fb62902706ae8e7aaff416b1b0fa2c0fd0878b46dc13baa3712d8a855" dependencies = [ - "base64 0.13.0", + "base64", "bitflags", - "bytes 1.0.1", + "bytes", "headers-core", "http", "mime", @@ -1567,7 +1564,7 @@ version = "0.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7245cd7449cc792608c3c8a9eaf69bd4eabbabf802713748fd739c98b82f0747" dependencies = [ - "bytes 1.0.1", + "bytes", "fnv", "itoa", ] @@ -1578,7 +1575,7 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2861bd27ee074e5ee891e8b539837a9430012e249d7f0ca2d795650f579c1994" dependencies = [ - "bytes 1.0.1", + "bytes", "http", ] @@ -1590,7 +1587,7 @@ checksum = "ad077d89137cd3debdce53c66714dc536525ef43fe075d41ddc0a8ac11f85957" dependencies = [ "anyhow", "async-channel", - "base64 0.13.0", + "base64", "futures-lite", "http", "infer", @@ -1621,7 +1618,7 @@ version = "0.14.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e8e946c2b1349055e0b72ae281b238baf1a3ea7307c7e9f9d64673bdd9c26ac7" dependencies = [ - "bytes 1.0.1", + "bytes", "futures-channel", "futures-core", "futures-util", @@ -1645,7 +1642,7 @@ version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6183ddfa99b85da61a140bea0efc93fdf56ceaa041b37d553518030827f9905" dependencies = [ - "bytes 1.0.1", + "bytes", "hyper", "native-tls", "tokio", @@ -1764,9 +1761,9 @@ dependencies = [ [[package]] name = "jsonrpc-core" -version = "17.1.0" +version = "18.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d4467ab6dfa369b69e52bd0692e480c4d117410538526a57a304a0f2250fd95e" +checksum = "14f7f76aef2d054868398427f6c54943cf3d1caa9a7ec7d0c38d69df97a965eb" dependencies = [ "futures", "futures-executor", @@ -2365,9 +2362,9 @@ dependencies = [ [[package]] name = "parse-display" -version = "0.5.0" +version = "0.5.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cc7e98ea043e0880940ef455c6c6e5710b4f670b4f0aeff6edf320bb01143fe9" +checksum = "898bf4c2a569dedbfd4e6c3f0bbd0ae825e5b6b0b69bae3e3c1000158689334a" dependencies = [ "once_cell", "parse-display-derive", @@ -2376,9 +2373,9 @@ dependencies = [ [[package]] name = "parse-display-derive" -version = "0.5.0" +version = "0.5.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "962e8dc54ebea1392eb2f36a205f2efa9437bfe8e95d7a91f070044c363c9684" +checksum = "1779d1e28ab04568223744c2af4aa4e642e67b92c76bdce0929a6d2c36267199" dependencies = [ "once_cell", "proc-macro2", @@ -2493,7 +2490,7 @@ version = "0.19.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0f853fba627ed1f21392d329eeb03caf90dce57a65dfbd24274f4c39452ed3bb" dependencies = [ - "bytes 1.0.1", + "bytes", "fallible-iterator", "futures", "log", @@ -2531,9 +2528,9 @@ version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ff3e0f70d32e20923cabf2df02913be7c1842d4c772db8065c00fcfdd1d1bff3" dependencies = [ - "base64 0.13.0", + "base64", "byteorder 1.4.3", - "bytes 1.0.1", + "bytes", "fallible-iterator", "hmac 0.10.1", "md-5", @@ -2549,7 +2546,7 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "430f4131e1b7657b0cd9a2b0c3408d77c9a43a042d300b8c77f981dffcc43a2f" dependencies = [ - "bytes 1.0.1", + "bytes", "chrono", "fallible-iterator", "postgres-derive", @@ -2566,9 +2563,9 @@ checksum = "ac74c624d6b2d21f425f752262f42188365d7b8ff1aff74c82e45136510a4857" [[package]] name = "pretty_assertions" -version = "0.7.2" +version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1cab0e7c02cf376875e9335e0ba1da535775beb5450d21e1dffca068818ed98b" +checksum = "ec0cfe1b2403f172ba0f234e500906ee0a3e493fb81092dac23ebefe129301cc" dependencies = [ "ansi_term 0.12.1", "ctor", @@ -2594,7 +2591,7 @@ name = "primitives" version = "0.1.0" dependencies = [ "async-trait", - "bytes 1.0.1", + "bytes", "chrono", "cid", "deadpool-postgres", @@ -2959,7 +2956,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d4f0ceb2ec0dd769483ecd283f6615aa83dcd0be556d5294c6e659caefe7cc54" dependencies = [ "async-trait", - "bytes 1.0.1", + "bytes", "combine", "dtoa", "futures-util", @@ -3023,8 +3020,8 @@ version = "0.11.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bf12057f289428dbf5c591c74bf10392e4a8003f993405a902f20117019022d4" dependencies = [ - "base64 0.13.0", - "bytes 1.0.1", + "base64", + "bytes", "encoding_rs", "futures-core", "futures-util", @@ -3083,7 +3080,7 @@ version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e54369147e3e7796c9b885c7304db87ca3d09a0a98f72843d532868675bbfba8" dependencies = [ - "bytes 1.0.1", + "bytes", "rustc-hex 2.1.0", ] @@ -3257,10 +3254,10 @@ dependencies = [ "dashmap", "deadpool 0.8.0", "deadpool-postgres", + "envy", "futures", "hex", "hyper", - "lazy_static", "migrant_lib", "once_cell", "postgres-types", @@ -3545,16 +3542,16 @@ dependencies = [ [[package]] name = "soketto" -version = "0.4.2" +version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b5c71ed3d54db0a699f4948e1bb3e45b450fa31fe602621dee6680361d569c88" +checksum = "4919971d141dbadaa0e82b5d369e2d7666c98e4625046140615ca363e50d4daa" dependencies = [ - "base64 0.12.3", - "bytes 0.5.6", + "base64", + "bytes", "futures", "httparse", "log", - "rand 0.7.3", + "rand 0.8.3", "sha-1", ] @@ -3609,9 +3606,9 @@ checksum = "73473c0e59e6d5812c5dfe2a064a6444949f089e20eec9a2e5506596494e4623" [[package]] name = "structmeta" -version = "0.1.3" +version = "0.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b55b4052fd036e3d1fe74ea978426a3f87997ba803e7a8e69ff0cf99f35a720a" +checksum = "59915b528a896f2e3bfa1a6ace65f7bb0ff9f9863de6213b0c01cb6fd3c3ac71" dependencies = [ "proc-macro2", "quote", @@ -3621,15 +3618,25 @@ dependencies = [ [[package]] name = "structmeta-derive" -version = "0.1.3" +version = "0.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3f55502dda4b5fd26b33f6810d7493b4f5d7859bca604bd07ff22a523cd257ee" +checksum = "b73800bcca56045d5ab138a48cd28a96093335335deaa916f22b5749c4150c79" dependencies = [ "proc-macro2", "quote", "syn", ] +[[package]] +name = "subprocess" +version = "0.2.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "055cf3ebc2981ad8f0a5a17ef6652f652d87831f79fddcba2ac57bcb9a0aa407" +dependencies = [ + "libc", + "winapi 0.3.9", +] + [[package]] name = "subtle" version = "2.4.0" @@ -3706,6 +3713,25 @@ dependencies = [ "winapi 0.3.9", ] +[[package]] +name = "test_harness" +version = "0.1.0" +dependencies = [ + "adapter", + "anyhow", + "chrono", + "futures", + "once_cell", + "primitives", + "reqwest", + "sentry", + "serde_json", + "slog", + "subprocess", + "tokio", + "web3", +] + [[package]] name = "textwrap" version = "0.11.0" @@ -3795,7 +3821,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2c2416fdedca8443ae44b4527de1ea633af61d8f7169ffa6e72c5b53d24efcc" dependencies = [ "autocfg 1.0.1", - "bytes 1.0.1", + "bytes", "libc", "memchr", "mio", @@ -3837,7 +3863,7 @@ checksum = "2d2b1383c7e4fb9a09e292c7c6afb7da54418d53b045f1c1fac7a911411a2b8b" dependencies = [ "async-trait", "byteorder 1.4.3", - "bytes 1.0.1", + "bytes", "fallible-iterator", "futures", "log", @@ -3869,7 +3895,7 @@ version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ec31e5cc6b46e653cf57762f36f71d5e6386391d88a72fd6db4508f8f676fb29" dependencies = [ - "bytes 1.0.1", + "bytes", "futures-core", "futures-io", "futures-sink", @@ -4023,7 +4049,6 @@ dependencies = [ "clap", "futures", "hex", - "lazy_static", "num", "num-traits", "primitives", @@ -4189,13 +4214,13 @@ dependencies = [ [[package]] name = "web3" -version = "0.16.0" +version = "0.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bc4c18ae15621f764fab919f7e4a83d87163494cbc3460884debef7c6bc1bc6b" +checksum = "cd24abe6f2b68e0677f843059faea87bcbd4892e39f02886f366d8222c3c540d" dependencies = [ "arrayvec 0.5.2", - "base64 0.13.0", - "bytes 1.0.1", + "base64", + "bytes", "derive_more", "ethabi", "ethereum-types 0.11.0", diff --git a/Cargo.toml b/Cargo.toml index 324ef2393..7ee561b02 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,4 +6,5 @@ members = [ "adview-manager", "validator_worker", "sentry", + "test_harness", ] diff --git a/adapter/Cargo.toml b/adapter/Cargo.toml index dde78180f..ff0f948e6 100644 --- a/adapter/Cargo.toml +++ b/adapter/Cargo.toml @@ -8,6 +8,10 @@ authors = [ ] edition = "2018" +[features] + +test-util = [] + [dependencies] primitives = { path = "../primitives" } # Time handling @@ -18,7 +22,7 @@ serde = { version = "^1.0", features = ['derive'] } serde_json = "1.0" serde-hex = "0.1.0" # Ethereum -web3 = { version = "0.16", features = ["http-tls", "signing"] } +web3 = { version = "0.17", features = ["http-tls", "signing"] } eth_checksum = "0.1" tiny-keccak = "1.5" ethstore = { git = "https://github.com/openethereum/openethereum", tag = "v3.1.1-rc.1" } @@ -29,7 +33,7 @@ reqwest = { version = "0.11", features = ["json"] } sha2 = "0.9" base64 = "0.13" -lazy_static = "^1.4" +once_cell = "^1.8" thiserror = "^1" # Futures futures = "0.3" diff --git a/adapter/Makefile.toml b/adapter/Makefile.toml index 0ca343a66..9a515f47c 100644 --- a/adapter/Makefile.toml +++ b/adapter/Makefile.toml @@ -14,9 +14,9 @@ dependencies = [ [tasks.ganache-up] script = ''' -docker-compose -f ../docker-compose.ci.yml up -d ganache \ +docker-compose -f ../docker-compose.harness.yml up --renew-anon-volumes -d ganache \ && sleep 10 ''' [tasks.ganache-down] -script = "docker-compose -f ../docker-compose.ci.yml down" +script = "docker-compose -f ../docker-compose.harness.yml down" diff --git a/adapter/src/dummy.rs b/adapter/src/dummy.rs index bf958b390..d4d30703a 100644 --- a/adapter/src/dummy.rs +++ b/adapter/src/dummy.rs @@ -172,7 +172,7 @@ impl Adapter for DummyAdapter { #[cfg(test)] mod test { use primitives::{ - config::configuration, + config::DEVELOPMENT_CONFIG, util::tests::prep_db::{ADDRESSES, DUMMY_CAMPAIGN, IDS}, BigNum, }; @@ -181,7 +181,7 @@ mod test { #[tokio::test] async fn test_deposits_calls() { - let config = configuration("development", None).expect("Should get Config"); + let config = DEVELOPMENT_CONFIG.clone(); let channel = DUMMY_CAMPAIGN.channel.clone(); let adapter = DummyAdapter::init( DummyAdapterOptions { diff --git a/adapter/src/ethereum.rs b/adapter/src/ethereum.rs index a7a53e7eb..d3247c88c 100644 --- a/adapter/src/ethereum.rs +++ b/adapter/src/ethereum.rs @@ -6,7 +6,7 @@ use ethstore::{ ethkey::{public_to_address, recover, verify_address, Message, Password, Signature}, SafeAccount, }; -use lazy_static::lazy_static; +use once_cell::sync::Lazy; use primitives::{ adapter::{Adapter, AdapterResult, Deposit, Error as AdapterError, KeystoreOptions, Session}, config::Config, @@ -26,27 +26,24 @@ use web3::{ Web3, }; -#[cfg(test)] -use test_utils::*; - mod error; -#[cfg(test)] -mod test_utils; +#[cfg(any(test, feature = "test-util"))] +pub mod test_util; -lazy_static! { - static ref OUTPACE_ABI: &'static [u8] = - include_bytes!("../../lib/protocol-eth/abi/OUTPACE.json"); - static ref ERC20_ABI: &'static [u8] = include_str!("../../lib/protocol-eth/abi/ERC20.json") +pub static OUTPACE_ABI: Lazy<&'static [u8]> = + Lazy::new(|| include_bytes!("../../lib/protocol-eth/abi/OUTPACE.json")); +pub static ERC20_ABI: Lazy<&'static [u8]> = Lazy::new(|| { + include_str!("../../lib/protocol-eth/abi/ERC20.json") .trim_end_matches('\n') - .as_bytes(); - static ref SWEEPER_ABI: &'static [u8] = - include_bytes!("../../lib/protocol-eth/abi/Sweeper.json"); - /// Ready to use init code (i.e. decoded) for calculating the create2 address - static ref DEPOSITOR_BYTECODE_DECODED: Vec = { - let bytecode = include_str!("../../lib/protocol-eth/resources/bytecode/Depositor.bin"); - hex::decode(bytecode).expect("Decoded properly") - }; -} + .as_bytes() +}); +pub static SWEEPER_ABI: Lazy<&'static [u8]> = + Lazy::new(|| include_bytes!("../../lib/protocol-eth/abi/Sweeper.json")); +/// Ready to use init code (i.e. decoded) for calculating the create2 address +pub static DEPOSITOR_BYTECODE_DECODED: Lazy> = Lazy::new(|| { + let bytecode = include_str!("../../lib/protocol-eth/resources/bytecode/Depositor.bin"); + hex::decode(bytecode).expect("Decoded properly") +}); trait EthereumChannel { fn tokenize(&self) -> Token; @@ -66,25 +63,25 @@ impl EthereumChannel for Channel { } } -fn get_counterfactual_address( - sweeper: H160, +pub fn get_counterfactual_address( + sweeper: Address, channel: &Channel, - outpace: H160, - depositor: &Address, -) -> H160 { + outpace: Address, + depositor: Address, +) -> Address { let salt: [u8; 32] = [0; 32]; let encoded_params = encode(&[ - Token::Address(outpace), + Token::Address(outpace.as_bytes().into()), channel.tokenize(), - Token::Address(H160(*depositor.as_bytes())), + Token::Address(depositor.as_bytes().into()), ]); let mut init_code = DEPOSITOR_BYTECODE_DECODED.clone(); init_code.extend(&encoded_params); - let address = calc_addr(sweeper.as_fixed_bytes(), &salt, &init_code); + let address_bytes = calc_addr(sweeper.as_bytes(), &salt, &init_code); - H160(address) + Address::from(address_bytes) } #[derive(Debug, Clone)] @@ -105,8 +102,9 @@ impl EthereumAdapter { let keystore_json: Value = serde_json::from_str(&keystore_contents).map_err(KeystoreError::Deserialization)?; - let address = keystore_json["address"] - .as_str() + let address = keystore_json + .get("address") + .and_then(|value| value.as_str()) .map(eth_checksum::checksum) .ok_or(KeystoreError::AddressMissing)?; @@ -134,13 +132,11 @@ impl Adapter for EthereumAdapter { type AdapterError = Error; fn unlock(&mut self) -> AdapterResult<(), Self::AdapterError> { - let account = SafeAccount::from_file( - serde_json::from_value(self.keystore_json.clone()) - .map_err(KeystoreError::Deserialization)?, - None, - &Some(self.keystore_pwd.clone()), - ) - .map_err(Error::WalletUnlock)?; + let json = serde_json::from_value(self.keystore_json.clone()) + .map_err(KeystoreError::Deserialization)?; + + let account = SafeAccount::from_file(json, None, &Some(self.keystore_pwd.clone())) + .map_err(Error::WalletUnlock)?; self.wallet = Some(account); @@ -270,6 +266,12 @@ impl Adapter for EthereumAdapter { channel: &Channel, depositor_address: &Address, ) -> AdapterResult { + let token_info = self + .config + .token_address_whitelist + .get(&channel.token) + .ok_or(Error::TokenNotWhitelisted(channel.token))?; + let outpace_contract = Contract::from_json( self.web3.eth(), self.config.outpace_address.into(), @@ -288,15 +290,15 @@ impl Adapter for EthereumAdapter { ) .map_err(Error::ContractInitialization)?; - let sweeper_address = sweeper_contract.address(); - let outpace_address = outpace_contract.address(); + let sweeper_address = Address::from(sweeper_contract.address().to_fixed_bytes()); + let outpace_address = Address::from(outpace_contract.address().to_fixed_bytes()); let on_outpace: U256 = outpace_contract .query( "deposits", ( Token::FixedBytes(channel.id().as_bytes().to_vec()), - Token::Address(depositor_address.as_bytes().into()), + Token::Address(H160(depositor_address.to_bytes())), ), None, Options::default(), @@ -311,12 +313,12 @@ impl Adapter for EthereumAdapter { sweeper_address, channel, outpace_address, - depositor_address, + *depositor_address, ); let still_on_create2: U256 = erc20_contract .query( "balanceOf", - counterfactual_address, + H160(counterfactual_address.to_bytes()), None, Options::default(), None, @@ -329,12 +331,6 @@ impl Adapter for EthereumAdapter { .parse() .map_err(Error::BigNumParsing)?; - let token_info = self - .config - .token_address_whitelist - .get(&channel.token) - .ok_or(Error::TokenNotWhitelisted(channel.token))?; - // Count the create2 deposit only if it's > minimum token units configured let deposit = if still_on_create2 > token_info.min_token_units_for_deposit { Deposit { @@ -502,8 +498,10 @@ pub fn ewt_verify( #[cfg(test)] mod test { + use super::test_util::*; use super::*; use chrono::Utc; + use primitives::{config::DEVELOPMENT_CONFIG, util::tests::prep_db::IDS}; use std::convert::TryFrom; use web3::{transports::Http, Web3}; use wiremock::{ @@ -513,14 +511,14 @@ mod test { #[test] fn should_init_and_unlock_ethereum_adapter() { - let mut eth_adapter = setup_eth_adapter(None, None, None); + let mut eth_adapter = setup_eth_adapter(DEVELOPMENT_CONFIG.clone()); eth_adapter.unlock().expect("should unlock eth adapter"); } #[test] fn should_get_whoami_sign_and_verify_messages() { // whoami - let mut eth_adapter = setup_eth_adapter(None, None, None); + let mut eth_adapter = setup_eth_adapter(DEVELOPMENT_CONFIG.clone()); let whoami = eth_adapter.whoami(); assert_eq!( whoami.to_string(), @@ -553,12 +551,7 @@ mod test { let message2 = "1648231285e69677531ffe70719f67a07f3d4393b8425a5a1c84b0c72434c77b"; let verify2 = eth_adapter - .verify( - ValidatorId::try_from("ce07CbB7e054514D590a0262C93070D838bFBA2e") - .expect("Failed to parse id"), - message2, - &signature2, - ) + .verify(IDS["leader"], message2, &signature2) .expect("Failed to verify signatures"); assert!(verify, "invalid signature 1 verification"); @@ -567,7 +560,7 @@ mod test { #[test] fn should_generate_correct_ewt_sign_and_verify() { - let mut eth_adapter = setup_eth_adapter(None, None, None); + let mut eth_adapter = setup_eth_adapter(DEVELOPMENT_CONFIG.clone()); eth_adapter.unlock().expect("should unlock eth adapter"); @@ -614,7 +607,7 @@ mod test { let mut identities_owned: HashMap = HashMap::new(); identities_owned.insert(identity, 2); - let mut eth_adapter = setup_eth_adapter(None, None, None); + let mut eth_adapter = setup_eth_adapter(DEVELOPMENT_CONFIG.clone()); Mock::given(method("GET")) .and(path(format!("/identity/by-owner/{}", eth_adapter.whoami()))) @@ -644,13 +637,13 @@ mod test { async fn get_deposit_and_count_create2_when_min_tokens_received() { let web3 = Web3::new(Http::new(&GANACHE_URL).expect("failed to init transport")); - let leader_account = H160(*GANACHE_ADDRESSES["leader"].as_bytes()); + let leader_account = GANACHE_ADDRESSES["leader"]; // deploy contracts let token = deploy_token_contract(&web3, 1_000) .await .expect("Correct parameters are passed to the Token constructor."); - let token_address = Address::from_bytes(&token.1.to_fixed_bytes()); + let token_address = token.1; let sweeper = deploy_sweeper_contract(&web3) .await @@ -664,15 +657,22 @@ mod test { let channel = get_test_channel(token_address); - let mut eth_adapter = setup_eth_adapter( - Some(*sweeper.0.as_fixed_bytes()), - Some(*outpace.0.as_fixed_bytes()), - Some((token_address, token.0)), + let mut config = DEVELOPMENT_CONFIG.clone(); + config.sweeper_address = sweeper.0.to_bytes(); + config.outpace_address = outpace.0.to_bytes(); + // since we deploy a new contract, it's should be different from all the ones found in config. + assert!( + config + .token_address_whitelist + .insert(token_address, token.0) + .is_none(), + "Should not have previous value, we've just deployed the contract." ); + let mut eth_adapter = setup_eth_adapter(config); eth_adapter.unlock().expect("should unlock eth adapter"); let counterfactual_address = - get_counterfactual_address(sweeper.0, &channel, outpace.0, &spender); + get_counterfactual_address(sweeper.0, &channel, outpace.0, spender); // No Regular nor Create2 deposits { @@ -696,14 +696,19 @@ mod test { &token.2, *GANACHE_ADDRESSES["leader"].as_bytes(), *spender.as_bytes(), - 10_000_u64, + &BigNum::from(10_000), ) .await .expect("Failed to set balance"); - outpace_deposit(&outpace.1, &channel, *spender.as_bytes(), 10_000) - .await - .expect("Should deposit funds"); + outpace_deposit( + &outpace.1, + &channel, + *spender.as_bytes(), + &BigNum::from(10_000), + ) + .await + .expect("Should deposit funds"); let regular_deposit = eth_adapter .get_deposit(&channel, &spender) @@ -724,9 +729,9 @@ mod test { // Set balance < minimal token units, i.e. `1_000` mock_set_balance( &token.2, - leader_account.to_fixed_bytes(), - counterfactual_address.to_fixed_bytes(), - 999_u64, + leader_account.to_bytes(), + counterfactual_address.to_bytes(), + &BigNum::from(999), ) .await .expect("Failed to set balance"); @@ -751,9 +756,9 @@ mod test { // Set balance > minimal token units mock_set_balance( &token.2, - leader_account.to_fixed_bytes(), - counterfactual_address.to_fixed_bytes(), - 1_999_u64, + leader_account.to_bytes(), + counterfactual_address.to_bytes(), + &BigNum::from(1_999), ) .await .expect("Failed to set balance"); @@ -777,9 +782,9 @@ mod test { { sweeper_sweep( &sweeper.1, - outpace.0.to_fixed_bytes(), + outpace.0.to_bytes(), &channel, - *spender.as_bytes(), + spender.to_bytes(), ) .await .expect("Should sweep the Spender account"); diff --git a/adapter/src/ethereum/test_util.rs b/adapter/src/ethereum/test_util.rs new file mode 100644 index 000000000..3c2efcc56 --- /dev/null +++ b/adapter/src/ethereum/test_util.rs @@ -0,0 +1,347 @@ +use once_cell::sync::Lazy; +use std::{collections::HashMap, convert::TryFrom, env::current_dir, num::NonZeroU8}; +use web3::{ + contract::{Contract, Options}, + ethabi::Token, + transports::Http, + types::{H160, H256, U256}, + Web3, +}; + +use primitives::{ + adapter::KeystoreOptions, + channel::{Channel, Nonce}, + config::TokenInfo, + Address, BigNum, Config, ValidatorId, +}; + +use crate::EthereumAdapter; + +use super::{EthereumChannel, OUTPACE_ABI, SWEEPER_ABI}; + +// See `adex-eth-protocol` `contracts/mocks/Token.sol` +/// Mocked Token ABI +pub static MOCK_TOKEN_ABI: Lazy<&'static [u8]> = + Lazy::new(|| include_bytes!("../../test/resources/mock_token_abi.json")); +/// Mocked Token bytecode in JSON +pub static MOCK_TOKEN_BYTECODE: Lazy<&'static str> = + Lazy::new(|| include_str!("../../test/resources/mock_token_bytecode.bin")); +/// Sweeper bytecode +pub static SWEEPER_BYTECODE: Lazy<&'static str> = + Lazy::new(|| include_str!("../../../lib/protocol-eth/resources/bytecode/Sweeper.bin")); +/// Outpace bytecode +pub static OUTPACE_BYTECODE: Lazy<&'static str> = + Lazy::new(|| include_str!("../../../lib/protocol-eth/resources/bytecode/OUTPACE.bin")); + +/// Uses local `keystore.json` file and it's address for testing and working with [`EthereumAdapter`] +pub static KEYSTORE_IDENTITY: Lazy<(Address, KeystoreOptions)> = Lazy::new(|| { + // The address of the keystore file in `adapter/test/resources/keystore.json` + let address = Address::try_from("0x2bDeAFAE53940669DaA6F519373f686c1f3d3393") + .expect("failed to parse id"); + + let full_path = current_dir().unwrap(); + // it always starts in `adapter` folder because of the crate scope + // even when it's in the workspace + let mut keystore_file = full_path.parent().unwrap().to_path_buf(); + keystore_file.push("adapter/test/resources/keystore.json"); + + (address, keystore_options("keystore.json", "adexvalidator")) +}); + +pub static GANACHE_KEYSTORES: Lazy> = Lazy::new(|| { + vec![ + ( + "guardian".to_string(), + ( + "0xDf08F82De32B8d460adbE8D72043E3a7e25A3B39" + .parse() + .expect("Valid Address"), + keystore_options( + "0xDf08F82De32B8d460adbE8D72043E3a7e25A3B39_keystore.json", + "address0", + ), + ), + ), + ( + "leader".to_string(), + ( + "0x5a04A8fB90242fB7E1db7d1F51e268A03b7f93A5" + .parse() + .expect("Valid Address"), + keystore_options( + "0x5a04A8fB90242fB7E1db7d1F51e268A03b7f93A5_keystore.json", + "address1", + ), + ), + ), + ( + "follower".to_string(), + ( + "0xe3896ebd3F32092AFC7D27e9ef7b67E26C49fB02" + .parse() + .expect("Valid Address"), + keystore_options( + "0xe3896ebd3F32092AFC7D27e9ef7b67E26C49fB02_keystore.json", + "address2", + ), + ), + ), + ( + "creator".to_string(), + ( + "0x0E45891a570Af9e5A962F181C219468A6C9EB4e1" + .parse() + .expect("Valid Address"), + keystore_options( + "0x0E45891a570Af9e5A962F181C219468A6C9EB4e1_keystore.json", + "address3", + ), + ), + ), + ( + "advertiser".to_string(), + ( + "0x8c4B95383a46D30F056aCe085D8f453fCF4Ed66d" + .parse() + .expect("Valid Address"), + keystore_options( + "0x8c4B95383a46D30F056aCe085D8f453fCF4Ed66d_keystore.json", + "address4", + ), + ), + ), + ( + "guardian2".to_string(), + ( + "0x1059B025E3F8b8f76A8120D6D6Fd9fBa172c80b8" + .parse() + .expect("Valid Address"), + keystore_options( + "0x1059B025E3F8b8f76A8120D6D6Fd9fBa172c80b8_keystore.json", + "address5", + ), + ), + ), + ] + .into_iter() + .collect() +}); + +/// Addresses generated on local running `ganache` for testing purposes. +/// see the `ganache-cli.sh` script in the repository +pub static GANACHE_ADDRESSES: Lazy> = Lazy::new(|| { + vec![ + ( + "guardian".to_string(), + "0xDf08F82De32B8d460adbE8D72043E3a7e25A3B39" + .parse() + .expect("Valid Address"), + ), + ( + "leader".to_string(), + "0x5a04A8fB90242fB7E1db7d1F51e268A03b7f93A5" + .parse() + .expect("Valid Address"), + ), + ( + "follower".to_string(), + "0xe3896ebd3F32092AFC7D27e9ef7b67E26C49fB02" + .parse() + .expect("Valid Address"), + ), + ( + "creator".to_string(), + "0x0E45891a570Af9e5A962F181C219468A6C9EB4e1" + .parse() + .expect("Valid Address"), + ), + ( + "advertiser".to_string(), + "0x8c4B95383a46D30F056aCe085D8f453fCF4Ed66d" + .parse() + .expect("Valid Address"), + ), + ( + "guardian2".to_string(), + "0x1059B025E3F8b8f76A8120D6D6Fd9fBa172c80b8" + .parse() + .expect("Valid Address"), + ), + ] + .into_iter() + .collect() +}); +/// Local `ganache` is running at: +pub const GANACHE_URL: &str = "http://localhost:8545"; + +/// This helper function generates the correct path to the keystore file from this file. +/// +/// The `file_name` located at `adapter/test/resources` +/// The `password` for the keystore file +fn keystore_options(file_name: &str, password: &str) -> KeystoreOptions { + let full_path = current_dir().unwrap(); + // it always starts in `adapter` folder because of the crate scope + // even when it's in the workspace + let mut keystore_file = full_path.parent().unwrap().to_path_buf(); + keystore_file.push(format!("adapter/test/resources/{}", file_name)); + + KeystoreOptions { + keystore_file: keystore_file.display().to_string(), + keystore_pwd: password.to_string(), + } +} + +pub fn get_test_channel(token_address: Address) -> Channel { + Channel { + leader: ValidatorId::from(&GANACHE_ADDRESSES["leader"]), + follower: ValidatorId::from(&GANACHE_ADDRESSES["follower"]), + guardian: GANACHE_ADDRESSES["advertiser"], + token: token_address, + nonce: Nonce::from(12345_u32), + } +} + +pub fn setup_eth_adapter(config: Config) -> EthereumAdapter { + EthereumAdapter::init(KEYSTORE_IDENTITY.1.clone(), &config) + .expect("should init ethereum adapter") +} + +pub async fn mock_set_balance( + token_contract: &Contract, + from: [u8; 20], + address: [u8; 20], + amount: &BigNum, +) -> web3::contract::Result { + let amount = U256::from_dec_str(&amount.to_string()).expect("Should create U256"); + + token_contract + .call( + "setBalanceTo", + (H160(address), amount), + H160(from), + Options::default(), + ) + .await +} + +pub async fn outpace_deposit( + outpace_contract: &Contract, + channel: &Channel, + to: [u8; 20], + amount: &BigNum, +) -> web3::contract::Result { + let amount = U256::from_dec_str(&amount.to_string()).expect("Should create U256"); + + outpace_contract + .call( + "deposit", + (channel.tokenize(), H160(to), amount), + H160(to), + Options::with(|opt| { + opt.gas_price = Some(1.into()); + opt.gas = Some(6_721_975.into()); + }), + ) + .await +} + +pub async fn sweeper_sweep( + sweeper_contract: &Contract, + outpace_address: [u8; 20], + channel: &Channel, + depositor: [u8; 20], +) -> web3::contract::Result { + let from_leader_account = H160(*GANACHE_ADDRESSES["leader"].as_bytes()); + + sweeper_contract + .call( + "sweep", + ( + Token::Address(H160(outpace_address)), + channel.tokenize(), + Token::Array(vec![Token::Address(H160(depositor))]), + ), + from_leader_account, + Options::with(|opt| { + opt.gas_price = Some(1.into()); + opt.gas = Some(6_721_975.into()); + }), + ) + .await +} + +/// Deploys the Sweeper contract from `GANACHE_ADDRESS['leader']` +pub async fn deploy_sweeper_contract( + web3: &Web3, +) -> web3::contract::Result<(Address, Contract)> { + let sweeper_contract = Contract::deploy(web3.eth(), &SWEEPER_ABI) + .expect("Invalid ABI of Sweeper contract") + .confirmations(0) + .options(Options::with(|opt| { + opt.gas_price = Some(1.into()); + opt.gas = Some(6_721_975.into()); + })) + .execute( + *SWEEPER_BYTECODE, + (), + H160(GANACHE_ADDRESSES["leader"].to_bytes()), + ) + .await?; + + let sweeper_address = Address::from(sweeper_contract.address().to_fixed_bytes()); + + Ok((sweeper_address, sweeper_contract)) +} + +/// Deploys the Outpace contract from `GANACHE_ADDRESS['leader']` +pub async fn deploy_outpace_contract( + web3: &Web3, +) -> web3::contract::Result<(Address, Contract)> { + let outpace_contract = Contract::deploy(web3.eth(), &OUTPACE_ABI) + .expect("Invalid ABI of Outpace contract") + .confirmations(0) + .options(Options::with(|opt| { + opt.gas_price = Some(1.into()); + opt.gas = Some(6_721_975.into()); + })) + .execute( + *OUTPACE_BYTECODE, + (), + H160(GANACHE_ADDRESSES["leader"].to_bytes()), + ) + .await?; + let outpace_address = Address::from(outpace_contract.address().to_fixed_bytes()); + + Ok((outpace_address, outpace_contract)) +} + +/// Deploys the Mock Token contract from `GANACHE_ADDRESS['leader']` +pub async fn deploy_token_contract( + web3: &Web3, + min_token_units: u64, +) -> web3::contract::Result<(TokenInfo, Address, Contract)> { + let token_contract = Contract::deploy(web3.eth(), &MOCK_TOKEN_ABI) + .expect("Invalid ABI of Mock Token contract") + .confirmations(0) + .options(Options::with(|opt| { + opt.gas_price = Some(1.into()); + opt.gas = Some(6_721_975.into()); + })) + .execute( + *MOCK_TOKEN_BYTECODE, + (), + H160(GANACHE_ADDRESSES["leader"].to_bytes()), + ) + .await?; + + let token_info = TokenInfo { + min_token_units_for_deposit: BigNum::from(min_token_units), + precision: NonZeroU8::new(18).expect("should create NonZeroU8"), + // 0.000_1 + min_validator_fee: BigNum::from(100_000_000_000_000), + }; + + let token_address = Address::from(token_contract.address().to_fixed_bytes()); + + Ok((token_info, token_address, token_contract)) +} diff --git a/adapter/src/ethereum/test_utils.rs b/adapter/src/ethereum/test_utils.rs deleted file mode 100644 index 4814e9b61..000000000 --- a/adapter/src/ethereum/test_utils.rs +++ /dev/null @@ -1,233 +0,0 @@ -use lazy_static::lazy_static; -use std::{collections::HashMap, num::NonZeroU8}; -use web3::{ - contract::{Contract, Options}, - ethabi::Token, - transports::Http, - types::{H160, H256, U256}, - Web3, -}; - -use primitives::{ - adapter::KeystoreOptions, - channel::{Channel, Nonce}, - config::{configuration, TokenInfo}, - Address, BigNum, ValidatorId, -}; - -use crate::EthereumAdapter; - -use super::{EthereumChannel, OUTPACE_ABI, SWEEPER_ABI}; - -// See `adex-eth-protocol` `contracts/mocks/Token.sol` -lazy_static! { - /// Mocked Token ABI - pub static ref MOCK_TOKEN_ABI: &'static [u8] = - include_bytes!("../../test/resources/mock_token_abi.json"); - /// Mocked Token bytecode in JSON - pub static ref MOCK_TOKEN_BYTECODE: &'static str = - include_str!("../../test/resources/mock_token_bytecode.bin"); - /// Sweeper bytecode - pub static ref SWEEPER_BYTECODE: &'static str = include_str!("../../../lib/protocol-eth/resources/bytecode/Sweeper.bin"); - /// Outpace bytecode - pub static ref OUTPACE_BYTECODE: &'static str = include_str!("../../../lib/protocol-eth/resources/bytecode/OUTPACE.bin"); - pub static ref GANACHE_ADDRESSES: HashMap = { - vec![ - ( - "leader".to_string(), - "0x5a04A8fB90242fB7E1db7d1F51e268A03b7f93A5" - .parse() - .expect("Valid Address"), - ), - ( - "follower".to_string(), - "0xe3896ebd3F32092AFC7D27e9ef7b67E26C49fB02" - .parse() - .expect("Valid Address"), - ), - ( - "creator".to_string(), - "0x0E45891a570Af9e5A962F181C219468A6C9EB4e1" - .parse() - .expect("Valid Address"), - ), - ( - "advertiser".to_string(), - "0x8c4B95383a46D30F056aCe085D8f453fCF4Ed66d" - .parse() - .expect("Valid Address"), - ), - ] - .into_iter() - .collect() - }; -} - -pub const GANACHE_URL: &'static str = "http://localhost:8545"; - -pub fn get_test_channel(token_address: Address) -> Channel { - Channel { - leader: ValidatorId::from(&GANACHE_ADDRESSES["leader"]), - follower: ValidatorId::from(&GANACHE_ADDRESSES["follower"]), - guardian: GANACHE_ADDRESSES["advertiser"], - token: token_address, - nonce: Nonce::from(12345_u32), - } -} - -pub fn setup_eth_adapter( - sweeper_address: Option<[u8; 20]>, - outpace_address: Option<[u8; 20]>, - token_whitelist: Option<(Address, TokenInfo)>, -) -> EthereumAdapter { - let mut config = configuration("development", None).expect("failed parse config"); - let keystore_options = KeystoreOptions { - keystore_file: "./test/resources/keystore.json".to_string(), - keystore_pwd: "adexvalidator".to_string(), - }; - - if let Some(address) = sweeper_address { - config.sweeper_address = address; - } - - if let Some(address) = outpace_address { - config.outpace_address = address; - } - - if let Some((address, token_info)) = token_whitelist { - assert!( - config - .token_address_whitelist - .insert(address, token_info) - .is_none(), - "It should not contain the generated token prior to this call!" - ) - } - - EthereumAdapter::init(keystore_options, &config).expect("should init ethereum adapter") -} - -pub async fn mock_set_balance( - token_contract: &Contract, - from: [u8; 20], - address: [u8; 20], - amount: u64, -) -> web3::contract::Result { - token_contract - .call( - "setBalanceTo", - (H160(address), U256::from(amount)), - H160(from), - Options::default(), - ) - .await -} - -pub async fn outpace_deposit( - outpace_contract: &Contract, - channel: &Channel, - to: [u8; 20], - amount: u64, -) -> web3::contract::Result { - outpace_contract - .call( - "deposit", - (channel.tokenize(), H160(to), U256::from(amount)), - H160(to), - Options::with(|opt| { - opt.gas_price = Some(1.into()); - opt.gas = Some(6_721_975.into()); - }), - ) - .await -} - -pub async fn sweeper_sweep( - sweeper_contract: &Contract, - outpace_address: [u8; 20], - channel: &Channel, - depositor: [u8; 20], -) -> web3::contract::Result { - let from_leader_account = H160(*GANACHE_ADDRESSES["leader"].as_bytes()); - - sweeper_contract - .call( - "sweep", - ( - Token::Address(H160(outpace_address)), - channel.tokenize(), - Token::Array(vec![Token::Address(H160(depositor))]), - ), - from_leader_account, - Options::with(|opt| { - opt.gas_price = Some(1.into()); - opt.gas = Some(6_721_975.into()); - }), - ) - .await -} - -/// Deploys the Sweeper contract from `GANACHE_ADDRESS['leader']` -pub async fn deploy_sweeper_contract( - web3: &Web3, -) -> web3::contract::Result<(H160, Contract)> { - let from_leader_account = H160(*GANACHE_ADDRESSES["leader"].as_bytes()); - - let sweeper_contract = Contract::deploy(web3.eth(), &SWEEPER_ABI) - .expect("Invalid ABI of Sweeper contract") - .confirmations(0) - .options(Options::with(|opt| { - opt.gas_price = Some(1.into()); - opt.gas = Some(6_721_975.into()); - })) - .execute(*SWEEPER_BYTECODE, (), from_leader_account) - .await?; - - Ok((sweeper_contract.address(), sweeper_contract)) -} - -/// Deploys the Outpace contract from `GANACHE_ADDRESS['leader']` -pub async fn deploy_outpace_contract( - web3: &Web3, -) -> web3::contract::Result<(H160, Contract)> { - let from_leader_account = H160(*GANACHE_ADDRESSES["leader"].as_bytes()); - - let outpace_contract = Contract::deploy(web3.eth(), &OUTPACE_ABI) - .expect("Invalid ABI of Sweeper contract") - .confirmations(0) - .options(Options::with(|opt| { - opt.gas_price = Some(1.into()); - opt.gas = Some(6_721_975.into()); - })) - .execute(*OUTPACE_BYTECODE, (), from_leader_account) - .await?; - - Ok((outpace_contract.address(), outpace_contract)) -} - -/// Deploys the Mock Token contract from `GANACHE_ADDRESS['leader']` -pub async fn deploy_token_contract( - web3: &Web3, - min_token_units: u64, -) -> web3::contract::Result<(TokenInfo, H160, Contract)> { - let from_leader_account = H160(*GANACHE_ADDRESSES["leader"].as_bytes()); - - let token_contract = Contract::deploy(web3.eth(), &MOCK_TOKEN_ABI) - .expect("Invalid ABI of Mock Token contract") - .confirmations(0) - .options(Options::with(|opt| { - opt.gas_price = Some(1.into()); - opt.gas = Some(6_721_975.into()); - })) - .execute(*MOCK_TOKEN_BYTECODE, (), from_leader_account) - .await?; - - let token_info = TokenInfo { - min_token_units_for_deposit: BigNum::from(min_token_units), - precision: NonZeroU8::new(18).expect("should create NonZeroU8"), - // 0.000_1 - min_validator_fee: BigNum::from(100_000_000_000_000), - }; - - Ok((token_info, token_contract.address(), token_contract)) -} diff --git a/adapter/test/resources/0x0E45891a570Af9e5A962F181C219468A6C9EB4e1_keystore.json b/adapter/test/resources/0x0E45891a570Af9e5A962F181C219468A6C9EB4e1_keystore.json new file mode 100644 index 000000000..4263a4890 --- /dev/null +++ b/adapter/test/resources/0x0E45891a570Af9e5A962F181C219468A6C9EB4e1_keystore.json @@ -0,0 +1 @@ +{"address":"0e45891a570af9e5a962f181c219468a6c9eb4e1","crypto":{"cipher":"aes-128-ctr","ciphertext":"639873e843cdf99a0c09cf24ee56569f35b4ebffeb054bf2b181f5c21545acee","cipherparams":{"iv":"281838b5a67842e79f8abe5cfb252567"},"kdf":"scrypt","kdfparams":{"dklen":32,"n":262144,"p":1,"r":8,"salt":"f2843ca233d543de6b545660b76f5a310ed30b50a0fbde7d6806d7fb53266711"},"mac":"9e73a152fcd224bfedf4625311ab625504a8a44f920d12bdc37884bead529941"},"id":"b757f415-d9f5-4c5e-b0ef-27a3b664d699","version":3} \ No newline at end of file diff --git a/adapter/test/resources/0x1059B025E3F8b8f76A8120D6D6Fd9fBa172c80b8_keystore.json b/adapter/test/resources/0x1059B025E3F8b8f76A8120D6D6Fd9fBa172c80b8_keystore.json new file mode 100644 index 000000000..f07b6bbb1 --- /dev/null +++ b/adapter/test/resources/0x1059B025E3F8b8f76A8120D6D6Fd9fBa172c80b8_keystore.json @@ -0,0 +1 @@ +{"address":"1059b025e3f8b8f76a8120d6d6fd9fba172c80b8","crypto":{"cipher":"aes-128-ctr","ciphertext":"16b6b4107693d7b607963c1c0a2be847453f9719087730bb56cb1d7da6769fd7","cipherparams":{"iv":"0e256dd2cb35ebb994c1e93ebe0ebc30"},"kdf":"scrypt","kdfparams":{"dklen":32,"n":262144,"p":1,"r":8,"salt":"cc99daa7de1b16c393c6a8689e8d0e22721e14389819076381e5a4c94f16cae2"},"mac":"fef83d9c3778abfff60804a75ec99c843a30b04fa1de5bc9124b784f7f42d88d"},"id":"d9fb8aee-951f-401d-aac7-8027fc16eed4","version":3} \ No newline at end of file diff --git a/adapter/test/resources/0x5a04A8fB90242fB7E1db7d1F51e268A03b7f93A5_keystore.json b/adapter/test/resources/0x5a04A8fB90242fB7E1db7d1F51e268A03b7f93A5_keystore.json new file mode 100644 index 000000000..d68345c40 --- /dev/null +++ b/adapter/test/resources/0x5a04A8fB90242fB7E1db7d1F51e268A03b7f93A5_keystore.json @@ -0,0 +1 @@ +{"address":"5a04a8fb90242fb7e1db7d1f51e268a03b7f93a5","crypto":{"cipher":"aes-128-ctr","ciphertext":"539dc1df190fbb6a148ede09174610a36ed507dba26679f6c9a6c242b2b8b952","cipherparams":{"iv":"bdd87ed2a71f3e5fb4ed89e2453ba011"},"kdf":"scrypt","kdfparams":{"dklen":32,"n":262144,"p":1,"r":8,"salt":"3f7a06ed550c2df365c02105de0f8de67bb54f4eb4ff442a541e9c9e34a51996"},"mac":"a5a2af6a3df54405a91686d2933498347b5719e71ee0bbb421a326ab3381ae27"},"id":"e52c9749-5653-4160-a87d-4cf23249eb44","version":3} \ No newline at end of file diff --git a/adapter/test/resources/0x8c4B95383a46D30F056aCe085D8f453fCF4Ed66d_keystore.json b/adapter/test/resources/0x8c4B95383a46D30F056aCe085D8f453fCF4Ed66d_keystore.json new file mode 100644 index 000000000..e62e47c08 --- /dev/null +++ b/adapter/test/resources/0x8c4B95383a46D30F056aCe085D8f453fCF4Ed66d_keystore.json @@ -0,0 +1 @@ +{"address":"8c4b95383a46d30f056ace085d8f453fcf4ed66d","crypto":{"cipher":"aes-128-ctr","ciphertext":"37efba60475fd119547800f72d7a5d4e5a189e9061ccb24ddef931a456c24b95","cipherparams":{"iv":"a55ceac53e55ddbb66646c3653286cc9"},"kdf":"scrypt","kdfparams":{"dklen":32,"n":262144,"p":1,"r":8,"salt":"5161d7bd1468517349c9c58866ea6bc425728b1bffc4e755b7b6bc225acefc65"},"mac":"125c30c396a08c01b60ebe57c955e87b44c138c9a398097858ac9d411f275536"},"id":"209c82fa-4300-49b0-a47e-3e5475423d58","version":3} \ No newline at end of file diff --git a/adapter/test/resources/0xDf08F82De32B8d460adbE8D72043E3a7e25A3B39_keystore.json b/adapter/test/resources/0xDf08F82De32B8d460adbE8D72043E3a7e25A3B39_keystore.json new file mode 100644 index 000000000..940ac1781 --- /dev/null +++ b/adapter/test/resources/0xDf08F82De32B8d460adbE8D72043E3a7e25A3B39_keystore.json @@ -0,0 +1 @@ +{"address":"df08f82de32b8d460adbe8d72043e3a7e25a3b39","crypto":{"cipher":"aes-128-ctr","ciphertext":"c4353fed51089663a658fdce24e9db24989061db6db131962a60046035ff37e3","cipherparams":{"iv":"8a4e5e8d9ac741e35b43de9420fb35aa"},"kdf":"scrypt","kdfparams":{"dklen":32,"n":262144,"p":1,"r":8,"salt":"6f7b6e17307cf067084baa0ba3ded8a47f410f02c642a8d674599df8378248db"},"mac":"8460fe570de1544df970493bb1c59b268343359ec43f5f93e7fa5b6ec9e3ea9c"},"id":"115ba830-f9c6-43c1-a3c2-9a5402ea542b","version":3} \ No newline at end of file diff --git a/adapter/test/resources/0xe3896ebd3F32092AFC7D27e9ef7b67E26C49fB02_keystore.json b/adapter/test/resources/0xe3896ebd3F32092AFC7D27e9ef7b67E26C49fB02_keystore.json new file mode 100644 index 000000000..c9cb1e803 --- /dev/null +++ b/adapter/test/resources/0xe3896ebd3F32092AFC7D27e9ef7b67E26C49fB02_keystore.json @@ -0,0 +1 @@ +{"address":"e3896ebd3f32092afc7d27e9ef7b67e26c49fb02","crypto":{"cipher":"aes-128-ctr","ciphertext":"372c9aaad3d866636c0e6c58c0669eddbccb6908caca3f1b81f7586dc36e595b","cipherparams":{"iv":"6d4bb1272873ec16671d68799544387d"},"kdf":"scrypt","kdfparams":{"dklen":32,"n":262144,"p":1,"r":8,"salt":"3a8bb6bb05fc14b389835439030284d70940e0e1fc5de4062542f97fdba9ac7b"},"mac":"ff7da3d71dc2abe47928189bfe1c906411c544f8e67b73bd0319329a743572b6"},"id":"f7f3eef3-bf10-416e-a5e7-e19ddeba1fe8","version":3} \ No newline at end of file diff --git a/adview-manager/Cargo.toml b/adview-manager/Cargo.toml index 503f2c24e..3699d79af 100644 --- a/adview-manager/Cargo.toml +++ b/adview-manager/Cargo.toml @@ -1,8 +1,5 @@ [package] -authors = [ - "Ambire ", - "Lachezar Lechev ", -] +authors = ["Ambire ", "Lachezar Lechev "] edition = "2018" name = "adview-manager" version = "0.1.0" diff --git a/docker-compose.ci.yml b/docker-compose.ci.yml index 14e8fe285..6cf01f119 100644 --- a/docker-compose.ci.yml +++ b/docker-compose.ci.yml @@ -10,8 +10,10 @@ services: - "5432:5432" # volumes: $HOME/docker/volumes/postgres:/var/lib/postgresql/data environment: - - POSTGRES_PASSWORD=postgres - - POSTGRES_DB=sentry_leader + POSTGRES_HOST: 'localhost' + POSTGRES_USER: 'postgres' + POSTGRES_PASSWORD: 'postgres' + POSTGRES_DB: 'sentry_leader' networks: - adex-leader - adex-external diff --git a/docker-compose.harness.yml b/docker-compose.harness.yml new file mode 100644 index 000000000..698842628 --- /dev/null +++ b/docker-compose.harness.yml @@ -0,0 +1,38 @@ +version: '3.8' + +services: + adex-postgres: + build: ./scripts/postgres + image: adex-postgres + container_name: adex-postgres + restart: always + ports: + - "5432:5432" + environment: + POSTGRES_HOST: 'localhost' + POSTGRES_USER: 'postgres' + POSTGRES_PASSWORD: 'postgres' + POSTGRES_MULTIPLE_DATABASES: harness_leader,harness_follower,sentry_leader,primitives + networks: + - adex-external + + adex-redis: + image: redis + container_name: adex-redis + restart: always + ports: + - "6379:6379" + networks: + - adex-external + ganache: + build: ./scripts/ethereum + image: adex-ganache + container_name: adex-ganache-cli + restart: always + ports: + - "8545:8545" + networks: + - adex-external + +networks: + adex-external: \ No newline at end of file diff --git a/docs/config/ganache.toml b/docs/config/ganache.toml new file mode 100644 index 000000000..0836b16a9 --- /dev/null +++ b/docs/config/ganache.toml @@ -0,0 +1,55 @@ +# based on: prod.toml +# Maximum number of channels to return per request +max_channels = 512 + +channels_find_limit = 512 +campaigns_find_limit = 512 +spendable_find_limit = 512 + +wait_time = 40000 + +# V4 Deprecated +aggr_throttle = 0 + +events_find_limit = 100 + +msgs_find_limit = 10 +analytics_find_limit_v5 = 5000 +analytics_maxtime_v5 = 15000 + +heartbeat_time = 60000 +health_threshold_promilles = 970 +health_unsignable_promilles = 770 +propagation_timeout = 3000 + +fetch_timeout = 10000 +all_campaigns_timeout = 10000 +channel_tick_timeout = 10000 + +ip_rate_limit = { type = 'ip', timeframe = 1200000 } +sid_rate_limit = { type = 'sid', timeframe = 0 } + +# Ganache Snapshot address +outpace_address = '0xcb097e455b7159f902e2eb45562fc397ae6b0f3d' +# Ganache Snapshot address +sweeper_address = '0xdd41b0069256a28972458199a3c9cf036384c156' + +ethereum_network = 'http://localhost:8545' +# Mocked relayer +# TODO: Remove #452 +ethereum_adapter_relayer = 'http://localhost:8888' + +creators_whitelist = [] +validators_whitelist = [] + + +[[token_address_whitelist]] +# Mocked TOKEN +address = '0x9db7bff788522dbe8fa2e8cbd568a58c471ccd5e' +# 1 * 10^18 = 1.0000 TOKEN +min_token_units_for_deposit = '1000000000000000000' +# multiplier = 10^14 - 10^18 (token precision) = 10^-4 +# min_validator_fee = 1 * 10^-4 = 0.000_1 +min_validator_fee = '100000000000000' +precision = 18 + diff --git a/primitives/Cargo.toml b/primitives/Cargo.toml index e30a5da33..e7e68e792 100644 --- a/primitives/Cargo.toml +++ b/primitives/Cargo.toml @@ -9,7 +9,7 @@ authors = [ edition = "2018" [features] -postgres = ["postgres-types", "bytes", "tokio-postgres"] +postgres = ["postgres-types", "bytes", "tokio-postgres", "deadpool-postgres"] [dependencies] # (De)Serialization @@ -50,14 +50,14 @@ num-derive = "0.3" fake = { version = "^1.3", features = ["chrono"] } rand = "^0.8" # postgres feature -postgres-types = { version = "0.2.0", features = [ - "with-serde_json-1", -], optional = true } +postgres-types = { version = "0.2.0", features = ["with-serde_json-1"], optional = true } bytes = { version = "^1", optional = true } tokio-postgres = { version = "0.7", optional = true, features = [ "with-chrono-0_4", "with-serde_json-1", ] } +# testing FromSql & ToSql implementation of structs +deadpool-postgres = { version = "0.9.0", optional = true } # Futures futures = "0.3" @@ -67,8 +67,6 @@ lazy_static = "1.4.0" once_cell = "^1.8" [dev-dependencies] -pretty_assertions = "^0.7" -# testing FromSql & ToSql implementation of structs -deadpool-postgres = "0.9.0" +pretty_assertions = "1" tokio = { version = "1", features = ["rt-multi-thread", "macros"] } serde_urlencoded = "^0.7" diff --git a/primitives/Makefile.toml b/primitives/Makefile.toml index fbd961cd5..fc0576e2c 100644 --- a/primitives/Makefile.toml +++ b/primitives/Makefile.toml @@ -13,10 +13,10 @@ dependencies = [ ] [tasks.test] -env = { "POSTGRES_DB" = "sentry_leader" } +env = { "POSTGRES_DB" = "primitives" } [tasks.services-up] -script = "docker-compose -f ../docker-compose.ci.yml up -d postgres-leader && sleep 2" +script = "docker-compose -f ../docker-compose.harness.yml up -d adex-postgres && sleep 6" [tasks.services-down] -script = "docker-compose -f ../docker-compose.ci.yml down" +script = "docker-compose -f ../docker-compose.harness.yml down" diff --git a/primitives/src/ad_slot.rs b/primitives/src/ad_slot.rs index e605488dd..ce5c16de8 100644 --- a/primitives/src/ad_slot.rs +++ b/primitives/src/ad_slot.rs @@ -7,6 +7,7 @@ use serde::{Deserialize, Serialize}; use std::collections::HashMap; /// See [AdEx Protocol adSlot.md][protocol] & [adex-models AdSlot.js][adex-models] for more details. +/// /// [protocol]: https://github.com/AdExNetwork/adex-protocol/blob/master/adSlot.md /// [adex-models]: https://github.com/AdExNetwork/adex-models/blob/master/src/models/AdSlot.js #[derive(Serialize, Deserialize, Debug, Clone, PartialEq)] diff --git a/primitives/src/address.rs b/primitives/src/address.rs index 7be07881f..87843ec99 100644 --- a/primitives/src/address.rs +++ b/primitives/src/address.rs @@ -26,6 +26,10 @@ pub struct Address( ); impl Address { + pub fn to_bytes(&self) -> [u8; 20] { + self.0 + } + pub fn as_bytes(&self) -> &[u8; 20] { &self.0 } @@ -65,12 +69,24 @@ impl From<&[u8; 20]> for Address { } } +impl From<[u8; 20]> for Address { + fn from(bytes: [u8; 20]) -> Self { + Self(bytes) + } +} + impl AsRef<[u8]> for Address { fn as_ref(&self) -> &[u8] { &self.0 } } +impl AsRef<[u8; 20]> for Address { + fn as_ref(&self) -> &[u8; 20] { + &self.0 + } +} + impl FromStr for Address { type Err = Error; diff --git a/primitives/src/big_num.rs b/primitives/src/big_num.rs index 1c11306ee..f370d1c6e 100644 --- a/primitives/src/big_num.rs +++ b/primitives/src/big_num.rs @@ -53,6 +53,26 @@ impl BigNum { pub fn from_bytes_be(buf: &[u8]) -> Self { Self(BigUint::from_bytes_be(buf)) } + + /// With this method you can easily create a [`BigNum`] from a whole number + /// + /// # Example + /// + /// ``` + /// # use primitives::BigNum; + /// let dai_precision = 18; + /// let whole_number = 15; + /// + /// let bignum = BigNum::with_precision(whole_number, dai_precision); + /// let expected = "15000000000000000000"; + /// + /// assert_eq!(expected, &bignum.to_string()); + /// ``` + pub fn with_precision(whole_number: u64, with_precision: u8) -> Self { + let multiplier = 10_u64.pow(with_precision.into()); + + BigNum::from(whole_number).mul(&multiplier) + } } impl fmt::Debug for BigNum { @@ -209,6 +229,15 @@ impl Mul<&BigNum> for BigNum { } } +impl Mul<&u64> for BigNum { + type Output = BigNum; + + fn mul(self, rhs: &u64) -> Self::Output { + let big_uint = &self.0 * rhs; + BigNum(big_uint) + } +} + impl<'a> Sum<&'a BigNum> for BigNum { fn sum>(iter: I) -> Self { let sum_uint = iter.map(|big_num| &big_num.0).sum(); diff --git a/primitives/src/campaign.rs b/primitives/src/campaign.rs index e7ab05545..1a7c818c8 100644 --- a/primitives/src/campaign.rs +++ b/primitives/src/campaign.rs @@ -278,6 +278,8 @@ mod pricing { } /// Campaign Validators pub mod validators { + use std::ops::Index; + use crate::{ValidatorDesc, ValidatorId}; use serde::{Deserialize, Serialize}; @@ -311,6 +313,17 @@ pub mod validators { } } + impl Index for Validators { + type Output = ValidatorDesc; + fn index(&self, index: usize) -> &Self::Output { + match index { + 0 => &self.0, + 1 => &self.1, + _ => panic!("Validators index is out of bound"), + } + } + } + /// Fixed size iterator of 2, as we need an iterator in couple of occasions impl<'a> IntoIterator for &'a Validators { type Item = &'a ValidatorDesc; diff --git a/primitives/src/channel.rs b/primitives/src/channel.rs index 4504add81..e78663fa2 100644 --- a/primitives/src/channel.rs +++ b/primitives/src/channel.rs @@ -375,7 +375,7 @@ pub mod postgres { #[cfg(test)] mod test { - use crate::{channel::Nonce, util::tests::prep_db::postgres::POSTGRES_POOL}; + use crate::{channel::Nonce, postgres::POSTGRES_POOL}; #[tokio::test] async fn nonce_to_from_sql() { let client = POSTGRES_POOL.get().await.unwrap(); diff --git a/primitives/src/config.rs b/primitives/src/config.rs index 074890911..62519dc73 100644 --- a/primitives/src/config.rs +++ b/primitives/src/config.rs @@ -2,17 +2,36 @@ use crate::{event_submission::RateLimit, Address, BigNum, ValidatorId}; use once_cell::sync::Lazy; use serde::{Deserialize, Deserializer, Serialize}; use serde_hex::{SerHex, StrictPfx}; -use std::{collections::HashMap, fs, num::NonZeroU8}; +use std::{collections::HashMap, num::NonZeroU8}; +use thiserror::Error; + +pub use toml::de::Error as TomlError; pub static DEVELOPMENT_CONFIG: Lazy = Lazy::new(|| { toml::from_str(include_str!("../../docs/config/dev.toml")) .expect("Failed to parse dev.toml config file") }); + pub static PRODUCTION_CONFIG: Lazy = Lazy::new(|| { toml::from_str(include_str!("../../docs/config/prod.toml")) .expect("Failed to parse prod.toml config file") }); +#[derive(Debug, Deserialize, PartialEq, Eq, Clone, Copy)] +#[serde(rename_all = "camelCase")] +/// The environment in which the application is running +/// Defaults to [`Environment::Development`] +pub enum Environment { + Development, + Production, +} + +impl Default for Environment { + fn default() -> Self { + Self::Development + } +} + #[derive(Serialize, Deserialize, Debug, Clone)] pub struct TokenInfo { pub min_token_units_for_deposit: BigNum, @@ -44,7 +63,7 @@ pub struct Config { /// In Milliseconds pub propagation_timeout: u32, /// in milliseconds - /// Set's the Client timeout for [`SentryApi`] + /// Set's the Client timeout for `SentryApi` /// This includes all requests made to sentry except propagating messages. /// When propagating messages we make requests to foreign Sentry instances as well. pub fetch_timeout: u32, @@ -66,6 +85,15 @@ pub struct Config { pub token_address_whitelist: HashMap, } +impl Config { + /// Utility method that will deserialize a Toml file content into a `Config`. + /// + /// Instead of relying on the `toml` crate directly, use this method instead. + pub fn try_toml(toml: &str) -> Result { + toml::from_str(toml) + } +} + #[derive(Serialize, Deserialize, Debug, Clone)] struct ConfigWhitelist { address: Address, @@ -99,26 +127,27 @@ where Ok(tokens_whitelist) } -#[derive(Serialize, Deserialize, Debug, Clone)] +#[derive(Debug, Error)] pub enum ConfigError { - InvalidFile(String), + #[error("Toml parsing: {0}")] + Toml(#[from] toml::de::Error), + #[error("File reading: {0}")] + InvalidFile(#[from] std::io::Error), } -pub fn configuration(environment: &str, config_file: Option<&str>) -> Result { +pub fn configuration( + environment: Environment, + config_file: Option<&str>, +) -> Result { match config_file { - Some(config_file) => match fs::read_to_string(config_file) { - Ok(config) => match toml::from_str(&config) { - Ok(data) => data, - Err(e) => Err(ConfigError::InvalidFile(e.to_string())), - }, - Err(e) => Err(ConfigError::InvalidFile(format!( - "Unable to read provided config file {} {}", - config_file, e - ))), - }, + Some(config_file) => { + let content = std::fs::read(config_file)?; + + Ok(toml::from_slice(&content)?) + } None => match environment { - "production" => Ok(PRODUCTION_CONFIG.clone()), - _ => Ok(DEVELOPMENT_CONFIG.clone()), + Environment::Production => Ok(PRODUCTION_CONFIG.clone()), + Environment::Development => Ok(DEVELOPMENT_CONFIG.clone()), }, } } diff --git a/primitives/src/lib.rs b/primitives/src/lib.rs index f448f882c..5352067e3 100644 --- a/primitives/src/lib.rs +++ b/primitives/src/lib.rs @@ -44,6 +44,81 @@ pub mod targeting; mod unified_num; pub mod validator; +/// This module is available with the `postgres` feature +/// Other places where you'd find `mod postgres` implementations is for many of the structs in the crate +/// all of which implement [`tokio_postres::FromSql`], [`tokio_postres::ToSql`] or [`From<&tokio_postgres::Row>`] +#[cfg(feature = "postgres")] +pub mod postgres { + use std::env::{self, VarError}; + + use deadpool_postgres::{Manager, ManagerConfig, Pool, RecyclingMethod}; + use once_cell::sync::Lazy; + use tokio_postgres::{Config, NoTls}; + + pub type DbPool = deadpool_postgres::Pool; + + /// A Postgres pool with reasonable settings: + /// - [`RecyclingMethod::Verified`] + /// - [`Pool::max_size`] = 32 + /// Created using environment variables, see [`POSTGRES_CONFIG`]. + pub static POSTGRES_POOL: Lazy = Lazy::new(|| { + let config = POSTGRES_CONFIG.clone(); + + let mgr_config = ManagerConfig { + recycling_method: RecyclingMethod::Verified, + }; + let mgr = Manager::from_config(config, NoTls, mgr_config); + + Pool::new(mgr, 42) + }); + + /// `POSTGRES_USER` environment variable - default: `postgres` + pub static POSTGRES_USER: Lazy = + Lazy::new(|| env::var("POSTGRES_USER").unwrap_or_else(|_| String::from("postgres"))); + + /// `POSTGRES_PASSWORD` environment variable - default: `postgres` + pub static POSTGRES_PASSWORD: Lazy = + Lazy::new(|| env::var("POSTGRES_PASSWORD").unwrap_or_else(|_| String::from("postgres"))); + + /// `POSTGRES_HOST` environment variable - default: `localhost` + pub static POSTGRES_HOST: Lazy = + Lazy::new(|| env::var("POSTGRES_HOST").unwrap_or_else(|_| String::from("localhost"))); + + /// `POSTGRES_PORT` environment variable - default: `5432` + pub static POSTGRES_PORT: Lazy = Lazy::new(|| { + env::var("POSTGRES_PORT") + .unwrap_or_else(|_| String::from("5432")) + .parse() + .unwrap() + }); + + /// `POSTGRES_DB` environment variable - default: `POSTGRES_USER` + pub static POSTGRES_DB: Lazy = Lazy::new(|| match env::var("POSTGRES_DB") { + Ok(database) => database, + Err(VarError::NotPresent) => POSTGRES_USER.clone(), + Err(err) => panic!("{}", err), + }); + + /// Postgres configuration derived from the environment variables: + /// - POSTGRES_USER + /// - POSTGRES_PASSWORD + /// - POSTGRES_HOST + /// - POSTGRES_PORT + /// - POSTGRES_DB + pub static POSTGRES_CONFIG: Lazy = Lazy::new(|| { + let mut config = Config::new(); + + config + .user(POSTGRES_USER.as_str()) + .password(POSTGRES_PASSWORD.as_str()) + .host(POSTGRES_HOST.as_str()) + .port(*POSTGRES_PORT) + .dbname(POSTGRES_DB.as_ref()); + + config + }); +} + mod deposit { use crate::{BigNum, UnifiedNum}; use serde::{Deserialize, Serialize}; diff --git a/primitives/src/sentry.rs b/primitives/src/sentry.rs index 6a62580dc..f1da99a32 100644 --- a/primitives/src/sentry.rs +++ b/primitives/src/sentry.rs @@ -227,6 +227,7 @@ pub struct AllSpendersResponse { #[derive(Debug, Serialize, Deserialize)] pub struct AllSpendersQuery { // default is `u64::default()` = `0` + #[serde(default)] pub page: u64, } @@ -440,9 +441,11 @@ pub mod campaign_create { }; #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] + #[serde(rename_all = "camelCase")] /// All fields are present except the `CampaignId` which is randomly created /// This struct defines the Body of the request (in JSON) pub struct CreateCampaign { + pub id: Option, pub channel: Channel, pub creator: Address, pub budget: UnifiedNum, @@ -470,10 +473,11 @@ pub mod campaign_create { } impl CreateCampaign { - /// Creates the new `Campaign` with randomly generated `CampaignId` + /// Creates a new [`Campaign`] + /// If [`CampaignId`] was not provided with the request it will be generated using [`CampaignId::new()`] pub fn into_campaign(self) -> Campaign { Campaign { - id: CampaignId::new(), + id: self.id.unwrap_or_else(CampaignId::new), channel: self.channel, creator: self.creator, budget: self.budget, @@ -487,13 +491,13 @@ pub mod campaign_create { active: self.active, } } - } - /// This implementation helps with test setup - /// **NOTE:** It erases the CampaignId, since the creation of the campaign gives it's CampaignId - impl From for CreateCampaign { - fn from(campaign: Campaign) -> Self { - Self { + /// Creates a [`CreateCampaign`] without using the [`Campaign.id`]. + /// You can either pass [`None`] to randomly generate a new [`CampaignId`]. + /// Or you can pass a [`CampaignId`] to be used for the [`CreateCampaign`]. + pub fn from_campaign_erased(campaign: Campaign, id: Option) -> Self { + CreateCampaign { + id, channel: campaign.channel, creator: campaign.creator, budget: campaign.budget, @@ -507,8 +511,35 @@ pub mod campaign_create { active: campaign.active, } } + + /// This function will retains the original [`Campaign.id`] ([`CampaignId`]). + pub fn from_campaign(campaign: Campaign) -> Self { + let id = Some(campaign.id); + Self::from_campaign_erased(campaign, id) + } } + // /// This implementation helps with test setup + // /// **NOTE:** It erases the CampaignId, since the creation of the campaign gives it's CampaignId + // impl From for CreateCampaign { + // fn from(campaign: Campaign) -> Self { + // Self { + // id: Some(campaign.id), + // channel: campaign.channel, + // creator: campaign.creator, + // budget: campaign.budget, + // validators: campaign.validators, + // title: campaign.title, + // pricing_bounds: campaign.pricing_bounds, + // event_submission: campaign.event_submission, + // ad_units: campaign.ad_units, + // targeting_rules: campaign.targeting_rules, + // created: campaign.created, + // active: campaign.active, + // } + // } + // } + // All editable fields stored in one place, used for checking when a budget is changed #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub struct ModifyCampaign { diff --git a/primitives/src/unified_num.rs b/primitives/src/unified_num.rs index 6dc91980b..7bae8b2e9 100644 --- a/primitives/src/unified_num.rs +++ b/primitives/src/unified_num.rs @@ -466,7 +466,7 @@ mod postgres { #[cfg(test)] mod test { use super::*; - use crate::util::tests::prep_db::postgres::POSTGRES_POOL; + use crate::postgres::POSTGRES_POOL; #[tokio::test] async fn from_and_to_sql() { diff --git a/primitives/src/util/logging.rs b/primitives/src/util/logging.rs index f1fda5de2..65019d2be 100644 --- a/primitives/src/util/logging.rs +++ b/primitives/src/util/logging.rs @@ -1,21 +1,31 @@ -use slog::{Drain, OwnedKVList, Record, KV}; +use slog::{o, Drain, Logger, OwnedKVList, Record, KV}; use slog_term::{ timestamp_local, CompactFormatSerializer, CountingWriter, Decorator, RecordDecorator, Serializer, ThreadSafeTimestampFn, }; -use std::cell::RefCell; -use std::{io, io::Write}; +use std::{ + cell::RefCell, + io::{Error, Result, Write}, +}; pub use slog_async::Async; pub use slog_term::TermDecorator; +pub fn new_logger(prefix: &str) -> Logger { + let decorator = TermDecorator::new().build(); + let drain = PrefixedCompactFormat::new(prefix, decorator).fuse(); + let drain = Async::new(drain).build().fuse(); + + Logger::root(drain, o!()) +} + pub struct PrefixedCompactFormat where D: Decorator, { decorator: D, history: RefCell, Vec)>>, - fn_timestamp: Box>>, + fn_timestamp: Box>>, prefix: String, } @@ -24,9 +34,9 @@ where D: Decorator, { type Ok = (); - type Err = io::Error; + type Err = Error; - fn log(&self, record: &Record<'_>, values: &OwnedKVList) -> Result { + fn log(&self, record: &Record<'_>, values: &OwnedKVList) -> Result { self.format_compact(record, values) } } @@ -44,7 +54,7 @@ where } } - fn format_compact(&self, record: &Record<'_>, values: &OwnedKVList) -> io::Result<()> { + fn format_compact(&self, record: &Record<'_>, values: &OwnedKVList) -> Result<()> { self.decorator.with_record(record, values, |decorator| { let indent = { let mut history_ref = self.history.borrow_mut(); @@ -83,10 +93,10 @@ where pub fn print_msg_header( prefix: &str, - fn_timestamp: &dyn ThreadSafeTimestampFn>, + fn_timestamp: &dyn ThreadSafeTimestampFn>, mut rd: &mut dyn RecordDecorator, record: &Record<'_>, -) -> io::Result { +) -> Result { rd.start_timestamp()?; fn_timestamp(&mut rd)?; diff --git a/primitives/src/util/tests/prep_db.rs b/primitives/src/util/tests/prep_db.rs index a2a7279c2..728fc0c6d 100644 --- a/primitives/src/util/tests/prep_db.rs +++ b/primitives/src/util/tests/prep_db.rs @@ -1,5 +1,5 @@ use crate::{ - campaign::{self, Active, Validators}, + campaign::{Active, Pricing, PricingBounds, Validators}, channel::Nonce, targeting::Rules, AdUnit, Address, Campaign, Channel, ChannelId, EventSubmission, UnifiedNum, ValidatorDesc, @@ -30,7 +30,7 @@ lazy_static! { pub static ref ADDRESSES: HashMap = { let mut addresses = HashMap::new(); - addresses.insert("leader".into(), Address::try_from("0xce07CbB7e054514D590a0262C93070D838bFBA2e").expect("failed to parse id")); + addresses.insert("leader".into(), Address::try_from("0xce07CbB7e054514D590a0262C93070D838bFBA2e").expect("failed to parse id")); addresses.insert("follower".into(), Address::try_from("0xc91763d7f14ac5c5ddfbcd012e0d2a61ab9bded3").expect("failed to parse id")); addresses.insert("user".into(), Address::try_from("0x20754168c00a6e58116ccfd0a5f7d1bb66c5de9d").expect("failed to parse id")); addresses.insert("publisher".into(), Address::try_from("0xb7d3f81e857692d13e9d63b232a90f4a1793189e").expect("failed to parse id")); @@ -96,7 +96,7 @@ lazy_static! { budget: UnifiedNum::from(100_000_000_000), validators: Validators::new((DUMMY_VALIDATOR_LEADER.clone(), DUMMY_VALIDATOR_FOLLOWER.clone())), title: Some("Dummy Campaign".to_string()), - pricing_bounds: Some(campaign::PricingBounds {impression: Some(campaign::Pricing { min: 1.into(), max: 10.into()}), click: Some(campaign::Pricing { min: 0.into(), max: 0.into()})}), + pricing_bounds: Some(PricingBounds {impression: Some(Pricing { min: 1.into(), max: 10.into()}), click: Some(Pricing { min: 0.into(), max: 0.into()})}), event_submission: Some(EventSubmission { allow: vec![] }), ad_units: vec![], targeting_rules: Rules::new(), @@ -188,48 +188,3 @@ lazy_static! { ]; } - -#[cfg(all(test, feature = "postgres"))] -pub mod postgres { - use deadpool_postgres::{Manager, ManagerConfig, Pool, RecyclingMethod}; - use lazy_static::lazy_static; - use once_cell::sync::Lazy; - use std::env; - use tokio_postgres::{Config, NoTls}; - - // TODO: Fix these values for usage in CI - lazy_static! { - static ref POSTGRES_USER: String = - env::var("POSTGRES_USER").unwrap_or_else(|_| String::from("postgres")); - static ref POSTGRES_PASSWORD: String = - env::var("POSTGRES_PASSWORD").unwrap_or_else(|_| String::from("postgres")); - static ref POSTGRES_HOST: String = - env::var("POSTGRES_HOST").unwrap_or_else(|_| String::from("localhost")); - static ref POSTGRES_PORT: u16 = env::var("POSTGRES_PORT") - .unwrap_or_else(|_| String::from("5432")) - .parse() - .unwrap(); - static ref POSTGRES_DB: Option = env::var("POSTGRES_DB").ok(); - } - - pub static POSTGRES_POOL: Lazy = Lazy::new(|| { - let mut config = Config::new(); - - config - .user(&POSTGRES_USER) - .password(POSTGRES_PASSWORD.as_str()) - .host(&POSTGRES_HOST) - .port(*POSTGRES_PORT); - - if let Some(db) = POSTGRES_DB.as_ref() { - config.dbname(db); - } - - let mgr_config = ManagerConfig { - recycling_method: RecyclingMethod::Fast, - }; - let mgr = Manager::from_config(config, NoTls, mgr_config); - let pool = Pool::new(mgr, 16); - pool - }); -} diff --git a/primitives/src/validator.rs b/primitives/src/validator.rs index 1b8f29592..ef6d208dc 100644 --- a/primitives/src/validator.rs +++ b/primitives/src/validator.rs @@ -2,7 +2,10 @@ use serde::{Deserialize, Serialize}; use std::{borrow::Borrow, convert::TryFrom, fmt, str::FromStr}; use crate::{ - address::Error, targeting::Value, Address, DomainError, ToETHChecksum, ToHex, UnifiedNum, + address::Error, + targeting::Value, + util::{api::Error as ApiUrlError, ApiUrl}, + Address, DomainError, ToETHChecksum, ToHex, UnifiedNum, }; pub use messages::*; @@ -96,17 +99,26 @@ impl TryFrom for ValidatorId { #[derive(Serialize, Deserialize, Debug, Clone, Eq, PartialEq)] #[serde(rename_all = "camelCase")] +/// A Validator description which includes the identity, fee (pro milles) and the Sentry URL. pub struct ValidatorDesc { pub id: ValidatorId, /// The validator fee in pro milles (per 1000) + /// Used to calculate the validator fee on each payout. pub fee: UnifiedNum, #[serde(default, skip_serializing_if = "Option::is_none")] /// The address which will receive the fees pub fee_addr: Option
, - /// The url of the Validator on which is the API + /// The url of the Validator where Sentry API is running pub url: String, } +impl ValidatorDesc { + /// Tries to create an [`ApiUrl`] from the `url` field. + pub fn try_api_url(&self) -> Result { + self.url.parse() + } +} + #[derive(Debug, PartialEq, Eq, Clone)] pub enum Validator { Leader(T), diff --git a/scripts/ethereum/ganache-cli.sh b/scripts/ethereum/ganache-cli.sh index 9f4cb94df..c58a6c867 100755 --- a/scripts/ethereum/ganache-cli.sh +++ b/scripts/ethereum/ganache-cli.sh @@ -3,12 +3,12 @@ # runs in Docker, so leave default port and export it instead of setting it up here -# Address 0: 0xDf08F82De32B8d460adbE8D72043E3a7e25A3B39 -# Address 1: 0x5a04A8fB90242fB7E1db7d1F51e268A03b7f93A5 -# Address 2: 0xe3896ebd3F32092AFC7D27e9ef7b67E26C49fB02 -# Address 3: 0x0E45891a570Af9e5A962F181C219468A6C9EB4e1 -# Address 4: 0x8c4B95383a46D30F056aCe085D8f453fCF4Ed66d -# Address 5: 0x1059B025E3F8b8f76A8120D6D6Fd9fBa172c80b8 +# Address 0: 0xDf08F82De32B8d460adbE8D72043E3a7e25A3B39 keystore password: address0 +# Address 1: 0x5a04A8fB90242fB7E1db7d1F51e268A03b7f93A5 keystore password: address1 +# Address 2: 0xe3896ebd3F32092AFC7D27e9ef7b67E26C49fB02 keystore password: address2 +# Address 3: 0x0E45891a570Af9e5A962F181C219468A6C9EB4e1 keystore password: address3 +# Address 4: 0x8c4B95383a46D30F056aCe085D8f453fCF4Ed66d keystore password: address4 +# Address 5: 0x1059B025E3F8b8f76A8120D6D6Fd9fBa172c80b8 keystore password: address5 node /app/ganache-core.docker.cli.js --gasLimit 0xfffffffffffff \ --db="./snapshot" \ --deterministic \ diff --git a/scripts/postgres/Dockerfile b/scripts/postgres/Dockerfile new file mode 100644 index 000000000..24a72c50a --- /dev/null +++ b/scripts/postgres/Dockerfile @@ -0,0 +1,5 @@ +FROM postgres:latest +COPY ./create-multiple-postgres-db.sh /docker-entrypoint-initdb.d + + +CMD ["docker-entrypoint.sh", "postgres"] \ No newline at end of file diff --git a/scripts/postgres/create-multiple-postgres-db.sh b/scripts/postgres/create-multiple-postgres-db.sh index 18c0f96b9..18ff3bd0e 100755 --- a/scripts/postgres/create-multiple-postgres-db.sh +++ b/scripts/postgres/create-multiple-postgres-db.sh @@ -4,19 +4,20 @@ set -e set -u function create_user_and_database() { - local database=$1 - echo " Creating user and database '$database'" - psql -v ON_ERROR_STOP=1 --username "$POSTGRES_USER" <<-EOSQL - CREATE USER $database; - CREATE DATABASE $database; - GRANT ALL PRIVILEGES ON DATABASE $database TO $database; + local database=$1 + + echo "Creating user and database '$database'" + psql -v ON_ERROR_STOP=1 --username "$POSTGRES_USER" <<-EOSQL + CREATE USER $database; + CREATE DATABASE $database; + GRANT ALL PRIVILEGES ON DATABASE $database TO $database; EOSQL } if [ -n "$POSTGRES_MULTIPLE_DATABASES" ]; then - echo "Multiple database creation requested: $POSTGRES_MULTIPLE_DATABASES" - for db in $(echo $POSTGRES_MULTIPLE_DATABASES | tr ',' ' '); do - create_user_and_database $db - done - echo "Multiple databases created" + echo "Multiple database creation requested: $POSTGRES_MULTIPLE_DATABASES" + for db in $(echo $POSTGRES_MULTIPLE_DATABASES | tr ',' ' '); do + create_user_and_database $db + done + echo "Multiple databases created" fi \ No newline at end of file diff --git a/sentry/Cargo.toml b/sentry/Cargo.toml index 3655c32fa..5624e055c 100644 --- a/sentry/Cargo.toml +++ b/sentry/Cargo.toml @@ -8,6 +8,10 @@ authors = [ ] edition = "2018" +[features] + +test-util = [] + [dependencies] # Futures futures = "^0.3" @@ -36,11 +40,12 @@ postgres-types = { version = "0.2.1", features = ["derive", "with-chrono-0_4", " migrant_lib = { version = "^0.32", features = ["d-postgres"] } # Logger slog = { version = "^2.2.3", features = ["max_level_trace"] } +# Deserialize values from Environment variables for the configuration +envy = "0.4" # Serde serde = { version = "^1.0", features = ["derive"] } serde_json = "^1.0" serde_urlencoded = "^0.7" # Other -lazy_static = "1.4.0" thiserror = "^1.0" once_cell = "1.5.2" diff --git a/sentry/Makefile.toml b/sentry/Makefile.toml index 24cf9bfcd..6a90fcff9 100644 --- a/sentry/Makefile.toml +++ b/sentry/Makefile.toml @@ -1,6 +1,3 @@ -[tasks.test] -env = { "POSTGRES_DB" = "sentry_leader" } - [tasks.dev-test-flow] description = "Development testing flow will first format the code, and than run cargo build and test" category = "Development" @@ -15,8 +12,11 @@ dependencies = [ "services-down", ] +[tasks.test] +env = { "POSTGRES_DB" = "sentry_leader" } + [tasks.services-up] -script = "docker-compose -f ../docker-compose.ci.yml up -d redis-leader postgres-leader && sleep 3" +script = "docker-compose -f ../docker-compose.harness.yml up -d adex-redis adex-postgres && sleep 3" [tasks.services-down] -script = "docker-compose -f ../docker-compose.ci.yml down" +script = "docker-compose -f ../docker-compose.harness.yml down" diff --git a/sentry/src/access.rs b/sentry/src/access.rs index d3435e88d..d6a970874 100644 --- a/sentry/src/access.rs +++ b/sentry/src/access.rs @@ -169,7 +169,7 @@ mod test { use chrono::TimeZone; use primitives::{ - config::configuration, + config::{configuration, Environment, DEVELOPMENT_CONFIG}, event_submission::{RateLimit, Rule}, sentry::Event, util::tests::prep_db::{ADDRESSES, DUMMY_CAMPAIGN, IDS}, @@ -187,7 +187,7 @@ mod test { async fn setup() -> (Config, Object) { let connection = TESTS_POOL.get().await.expect("Should return Object"); - let config = configuration("development", None).expect("Failed to get dev configuration"); + let config = DEVELOPMENT_CONFIG.clone(); (config, connection) } diff --git a/sentry/src/application.rs b/sentry/src/application.rs new file mode 100644 index 000000000..019aab4c7 --- /dev/null +++ b/sentry/src/application.rs @@ -0,0 +1,120 @@ +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; + +use hyper::{ + service::{make_service_fn, service_fn}, + Error, Server, +}; +use once_cell::sync::Lazy; +use primitives::{adapter::Adapter, config::Environment}; +use redis::ConnectionInfo; +use serde::{Deserialize, Deserializer}; +use slog::{error, info, Logger}; + +/// an error used when deserializing a [`Config`] instance from environment variables +/// see [`Config::from_env()`] +pub use envy::Error as EnvError; + +use crate::Application; + +pub const DEFAULT_PORT: u16 = 8005; +pub const DEFAULT_IP_ADDR: IpAddr = IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)); +pub static DEFAULT_REDIS_URL: Lazy = Lazy::new(|| { + "redis://127.0.0.1:6379" + .parse::() + .expect("Valid URL") +}); +#[derive(Debug, Deserialize, Clone)] +pub struct Config { + /// Defaults to `Development`: [`Environment::default()`] + pub env: Environment, + /// The port on which the Sentry REST API will be accessible. + #[serde(default = "default_port")] + /// Defaults to `8005`: [`DEFAULT_PORT`] + pub port: u16, + /// The address on which the Sentry REST API will be accessible. + /// `0.0.0.0` can be used for Docker. + /// `127.0.0.1` can be used for locally running servers. + #[serde(default = "default_ip_addr")] + /// Defaults to `0.0.0.0`: [`DEFAULT_IP_ADDR`] + pub ip_addr: IpAddr, + #[serde(deserialize_with = "redis_url", default = "default_redis_url")] + /// Defaults to locally running Redis server: [`DEFAULT_REDIS_URL`] + pub redis_url: ConnectionInfo, +} + +impl Config { + /// Deserialize the application [`Config`] from Environment variables. + pub fn from_env() -> Result { + envy::from_env() + } +} + +fn redis_url<'a, 'de: 'a, D>(deserializer: D) -> Result +where + D: Deserializer<'de>, +{ + let url_string = <&'a str>::deserialize(deserializer)?; + + url_string.parse().map_err(serde::de::Error::custom) +} + +fn default_port() -> u16 { + DEFAULT_PORT +} +fn default_ip_addr() -> IpAddr { + DEFAULT_IP_ADDR +} +fn default_redis_url() -> ConnectionInfo { + DEFAULT_REDIS_URL.clone() +} + +/// Starts the `hyper` `Server`. +pub async fn run(app: Application, socket_addr: SocketAddr) { + let logger = app.logger.clone(); + info!(&logger, "Listening on socket address: {}!", socket_addr); + + let make_service = make_service_fn(|_| { + let server = app.clone(); + async move { + Ok::<_, Error>(service_fn(move |req| { + let server = server.clone(); + async move { Ok::<_, Error>(server.handle_routing(req).await) } + })) + } + }); + + let server = Server::bind(&socket_addr).serve(make_service); + + if let Err(e) = server.await { + error!(&logger, "server error: {}", e; "main" => "run"); + } +} + +pub fn logger(prefix: &str) -> Logger { + use primitives::util::logging::{Async, PrefixedCompactFormat, TermDecorator}; + use slog::{o, Drain}; + + let decorator = TermDecorator::new().build(); + let drain = PrefixedCompactFormat::new(prefix, decorator).fuse(); + let drain = Async::new(drain).build().fuse(); + + Logger::root(drain, o!()) +} + +#[cfg(test)] +mod test { + use serde_json::json; + + use super::*; + + #[test] + fn environment() { + let development = serde_json::from_value::(json!("development")) + .expect("Should deserialize"); + let production = + serde_json::from_value::(json!("production")).expect("Should deserialize"); + + assert_eq!(Environment::Development, development); + assert_eq!(Environment::Production, production); + } +} diff --git a/sentry/src/db.rs b/sentry/src/db.rs index b0656ec1d..9b64fcac3 100644 --- a/sentry/src/db.rs +++ b/sentry/src/db.rs @@ -1,13 +1,15 @@ use deadpool_postgres::{Manager, ManagerConfig, RecyclingMethod}; -use redis::aio::MultiplexedConnection; -use std::{env, str::FromStr}; +use primitives::{ + config::Environment, + postgres::{POSTGRES_DB, POSTGRES_HOST, POSTGRES_PASSWORD, POSTGRES_PORT, POSTGRES_USER}, +}; +use redis::{aio::MultiplexedConnection, IntoConnectionInfo}; +use std::str::FromStr; use tokio_postgres::{ types::{accepts, FromSql, Type}, NoTls, }; -use lazy_static::lazy_static; - pub mod accounting; pub mod analytics; pub mod campaign; @@ -21,6 +23,9 @@ pub use self::channel::*; pub use self::event_aggregate::*; pub use self::validator_message::*; +// Re-export the Postgres Config +pub use tokio_postgres::Config as PostgresConfig; + // Re-export the Postgres PoolError for easier usages pub use deadpool_postgres::PoolError; // Re-export the redis RedisError for easier usage @@ -28,34 +33,6 @@ pub use redis::RedisError; pub type DbPool = deadpool_postgres::Pool; -lazy_static! { - static ref POSTGRES_USER: String = - env::var("POSTGRES_USER").unwrap_or_else(|_| String::from("postgres")); - static ref POSTGRES_PASSWORD: String = - env::var("POSTGRES_PASSWORD").unwrap_or_else(|_| String::from("postgres")); - static ref POSTGRES_HOST: String = - env::var("POSTGRES_HOST").unwrap_or_else(|_| String::from("localhost")); - static ref POSTGRES_PORT: u16 = env::var("POSTGRES_PORT") - .unwrap_or_else(|_| String::from("5432")) - .parse() - .unwrap(); - static ref POSTGRES_DB: Option = env::var("POSTGRES_DB").ok(); - static ref POSTGRES_CONFIG: tokio_postgres::Config = { - let mut config = tokio_postgres::Config::new(); - - config - .user(POSTGRES_USER.as_str()) - .password(POSTGRES_PASSWORD.as_str()) - .host(POSTGRES_HOST.as_str()) - .port(*POSTGRES_PORT); - if let Some(db) = POSTGRES_DB.as_ref() { - config.dbname(db); - } - - config - }; -} - pub struct TotalCount(pub u64); impl<'a> FromSql<'a> for TotalCount { fn from_sql( @@ -71,23 +48,26 @@ impl<'a> FromSql<'a> for TotalCount { accepts!(VARCHAR, TEXT); } -pub async fn redis_connection(url: &str) -> Result { +pub async fn redis_connection( + url: impl IntoConnectionInfo, +) -> Result { let client = redis::Client::open(url)?; client.get_multiplexed_async_connection().await } -pub async fn postgres_connection(max_size: usize) -> DbPool { +pub async fn postgres_connection(max_size: usize, config: tokio_postgres::Config) -> DbPool { let mgr_config = ManagerConfig { recycling_method: RecyclingMethod::Verified, }; - let manager = Manager::from_config(POSTGRES_CONFIG.clone(), NoTls, mgr_config); + let manager = Manager::from_config(config, NoTls, mgr_config); DbPool::new(manager, max_size) } -pub async fn setup_migrations(environment: &str) { +/// Sets the migrations using the `POSTGRES_*` environment variables +pub async fn setup_migrations(environment: Environment) { use migrant_lib::{Config, Direction, Migrator, Settings}; let settings = Settings::configure_postgres() @@ -95,7 +75,7 @@ pub async fn setup_migrations(environment: &str) { .database_password(POSTGRES_PASSWORD.as_str()) .database_host(POSTGRES_HOST.as_str()) .database_port(*POSTGRES_PORT) - .database_name(POSTGRES_DB.as_ref().unwrap_or(&POSTGRES_USER)) + .database_name(POSTGRES_DB.as_ref()) .build() .expect("Should build migration settings"); @@ -118,7 +98,7 @@ pub async fn setup_migrations(environment: &str) { // `tests_postgres::MIGRATIONS` let mut migrations = vec![make_migration!("20190806011140_initial-tables")]; - if environment == "development" { + if let Environment::Development = environment { // seeds database tables for testing migrations.push(make_migration!("20190806011140_initial-tables/seed")); } @@ -131,7 +111,7 @@ pub async fn setup_migrations(environment: &str) { // Reload config, ping the database for applied migrations let config = config.reload().expect("Should reload applied migrations"); - if environment == "development" { + if let Environment::Development = environment { // delete all existing data to make tests reproducible Migrator::with_config(&config) .all(true) @@ -158,7 +138,7 @@ pub async fn setup_migrations(environment: &str) { .expect("Reloading config for migration failed"); } -#[cfg(test)] +#[cfg(feature = "test-util")] pub mod tests_postgres { use std::{ ops::{Deref, DerefMut}, @@ -168,11 +148,12 @@ pub mod tests_postgres { use deadpool::managed::{Manager as ManagerTrait, RecycleResult}; use deadpool_postgres::ManagerConfig; use once_cell::sync::Lazy; + use primitives::postgres::POSTGRES_CONFIG; use tokio_postgres::{NoTls, SimpleQueryMessage}; use async_trait::async_trait; - use super::{DbPool, PoolError, POSTGRES_CONFIG}; + use super::{DbPool, PoolError}; pub type Pool = deadpool::managed::Pool; @@ -348,10 +329,18 @@ pub mod tests_postgres { // DROP the public schema and create it again for usage after recycling let queries = "DROP SCHEMA public CASCADE; CREATE SCHEMA public;"; - let result = database - .pool - .get() - .await? + database.pool = { + let mut config = self.base_config.clone(); + // set the database in the configuration of the inside Pool (used for tests) + config.dbname(&database.name); + + let manager = + deadpool_postgres::Manager::from_config(config, NoTls, self.manager_config.clone()); + + deadpool_postgres::Pool::new(manager, 15) + }; + + let result = database.pool.get().await? .simple_query(queries) .await .map_err(PoolError::Backend) @@ -370,19 +359,15 @@ pub mod tests_postgres { let full_query: String = MIGRATIONS .iter() .map(|migration| { - use std::{ - fs::File, - io::{BufReader, Read}, - }; - let file = File::open(format!("migrations/{}/up.sql", migration)) - .expect("File migration couldn't be opened"); - let mut buf_reader = BufReader::new(file); - let mut contents = String::new(); - - buf_reader - .read_to_string(&mut contents) - .expect("File migration couldn't be read"); - contents + use std::{env::current_dir, fs::read_to_string}; + + let full_path = current_dir().unwrap(); + // it always starts in `sentry` folder because of the crate scope + // even when it's in the workspace + let mut file = full_path.parent().unwrap().to_path_buf(); + file.push(format!("sentry/migrations/{}/up.sql", migration)); + + read_to_string(file).expect("File migration couldn't be read") }) .collect(); @@ -443,7 +428,7 @@ pub mod tests_postgres { } } -#[cfg(test)] +#[cfg(feature = "test-util")] pub mod redis_pool { use dashmap::DashMap; @@ -457,6 +442,9 @@ pub mod redis_pool { use super::*; + /// Re-export [`redis::cmd`] for testing purposes + pub use redis::cmd; + pub type Pool = deadpool::managed::Pool; pub static TESTS_POOL: Lazy = @@ -546,7 +534,7 @@ pub mod redis_pool { // see https://github.com/mitsuhiko/redis-rs/issues/325 _ => { let mut redis_conn = - redis_connection(&format!("{}{}", Self::URL, record.key())) + redis_connection(format!("{}{}", Self::URL, record.key())) .await .expect("Should connect"); @@ -574,7 +562,7 @@ pub mod redis_pool { async fn recycle(&self, database: &mut Database) -> RecycleResult { // always make a new connection because of know redis crate issue // see https://github.com/mitsuhiko/redis-rs/issues/325 - let connection = redis_connection(&format!("{}{}", Self::URL, database.index)) + let connection = redis_connection(format!("{}{}", Self::URL, database.index)) .await .expect("Should connect"); // make the database available diff --git a/sentry/src/lib.rs b/sentry/src/lib.rs index 6e1f7ef2d..a983a30fa 100644 --- a/sentry/src/lib.rs +++ b/sentry/src/lib.rs @@ -4,7 +4,6 @@ use chrono::Utc; use hyper::{Body, Method, Request, Response, StatusCode}; -use lazy_static::lazy_static; use middleware::{ auth::{AuthRequired, Authenticate}, campaign::{CalledByCreator, CampaignLoad}, @@ -45,6 +44,7 @@ pub mod routes { pub mod access; pub mod analytics_recorder; +pub mod application; pub mod db; // TODO AIP#61: remove the even aggregator once we've taken out the logic for AIP#61 // pub mod event_aggregator; @@ -53,20 +53,34 @@ pub mod db; pub mod payout; pub mod spender; -lazy_static! { - static ref CHANNEL_GET_BY_ID: Regex = - Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/?$").expect("The regex should be valid"); - static ref LAST_APPROVED_BY_CHANNEL_ID: Regex = Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/last-approved/?$").expect("The regex should be valid"); - static ref CHANNEL_STATUS_BY_CHANNEL_ID: Regex = Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/status/?$").expect("The regex should be valid"); - // Only the initial Regex to be matched. - static ref CHANNEL_VALIDATOR_MESSAGES: Regex = Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/validator-messages(/.*)?$").expect("The regex should be valid"); - static ref CHANNEL_EVENTS_AGGREGATES: Regex = Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/events-aggregates/?$").expect("The regex should be valid"); - static ref ANALYTICS_BY_CHANNEL_ID: Regex = Regex::new(r"^/analytics/0x([a-zA-Z0-9]{64})/?$").expect("The regex should be valid"); - static ref ADVERTISER_ANALYTICS_BY_CHANNEL_ID: Regex = Regex::new(r"^/analytics/for-advertiser/0x([a-zA-Z0-9]{64})/?$").expect("The regex should be valid"); - static ref PUBLISHER_ANALYTICS_BY_CHANNEL_ID: Regex = Regex::new(r"^/analytics/for-publisher/0x([a-zA-Z0-9]{64})/?$").expect("The regex should be valid"); - static ref CREATE_EVENTS_BY_CHANNEL_ID: Regex = Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/events/?$").expect("The regex should be valid"); - static ref CHANNEL_SPENDER_LEAF_AND_TOTAL_DEPOSITED: Regex = Regex::new(r"^/v5/channel/0x([a-zA-Z0-9]{64})/spender/0x([a-zA-Z0-9]{40})/?$").expect("This regex should be valid"); -} +static LAST_APPROVED_BY_CHANNEL_ID: Lazy = Lazy::new(|| { + Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/last-approved/?$") + .expect("The regex should be valid") +}); +// Only the initial Regex to be matched. +static CHANNEL_VALIDATOR_MESSAGES: Lazy = Lazy::new(|| { + Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/validator-messages(/.*)?$") + .expect("The regex should be valid") +}); +static CHANNEL_EVENTS_AGGREGATES: Lazy = Lazy::new(|| { + Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/events-aggregates/?$") + .expect("The regex should be valid") +}); +static ANALYTICS_BY_CHANNEL_ID: Lazy = Lazy::new(|| { + Regex::new(r"^/analytics/0x([a-zA-Z0-9]{64})/?$").expect("The regex should be valid") +}); +static ADVERTISER_ANALYTICS_BY_CHANNEL_ID: Lazy = Lazy::new(|| { + Regex::new(r"^/analytics/for-advertiser/0x([a-zA-Z0-9]{64})/?$") + .expect("The regex should be valid") +}); +static PUBLISHER_ANALYTICS_BY_CHANNEL_ID: Lazy = Lazy::new(|| { + Regex::new(r"^/analytics/for-publisher/0x([a-zA-Z0-9]{64})/?$") + .expect("The regex should be valid") +}); +static CHANNEL_SPENDER_LEAF_AND_TOTAL_DEPOSITED: Lazy = Lazy::new(|| { + Regex::new(r"^/v5/channel/0x([a-zA-Z0-9]{64})/spender/0x([a-zA-Z0-9]{40})/?$") + .expect("This regex should be valid") +}); static INSERT_EVENTS_BY_CAMPAIGN_ID: Lazy = Lazy::new(|| { Regex::new(r"^/v5/campaign/0x([a-zA-Z0-9]{32})/events/?$").expect("The regex should be valid") @@ -548,7 +562,7 @@ pub mod test_util { use adapter::DummyAdapter; use primitives::{ adapter::DummyAdapterOptions, - config::configuration, + config::DEVELOPMENT_CONFIG, util::tests::{discard_logger, prep_db::IDS}, }; @@ -561,9 +575,10 @@ pub mod test_util { Application, }; - /// Uses development and therefore the goreli testnet addresses of the tokens + /// Uses development and therefore the goerli testnet addresses of the tokens + /// It still uses DummyAdapter. pub async fn setup_dummy_app() -> Application { - let config = configuration("development", None).expect("Should get Config"); + let config = DEVELOPMENT_CONFIG.clone(); let adapter = DummyAdapter::init( DummyAdapterOptions { dummy_identity: IDS["leader"], diff --git a/sentry/src/main.rs b/sentry/src/main.rs index 0ab25891c..20afd5c5a 100644 --- a/sentry/src/main.rs +++ b/sentry/src/main.rs @@ -4,23 +4,20 @@ use clap::{crate_version, App, Arg}; use adapter::{AdapterTypes, DummyAdapter, EthereumAdapter}; -use hyper::service::{make_service_fn, service_fn}; -use hyper::{Error, Server}; -use primitives::adapter::{Adapter, DummyAdapterOptions, KeystoreOptions}; -use primitives::config::configuration; -use primitives::util::tests::prep_db::{AUTH, IDS}; -use primitives::ValidatorId; -use sentry::db::{postgres_connection, redis_connection, setup_migrations, CampaignRemaining}; -use sentry::Application; -use slog::{error, info, Logger}; -use std::{ - convert::TryFrom, - env, - net::{IpAddr, Ipv4Addr, SocketAddr}, +use primitives::{ + adapter::{DummyAdapterOptions, KeystoreOptions}, + config::configuration, + postgres::POSTGRES_CONFIG, + util::tests::prep_db::{AUTH, IDS}, + ValidatorId, }; - -const DEFAULT_PORT: u16 = 8005; -const DEFAULT_IP_ADDR: IpAddr = IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)); +use sentry::{ + application::{logger, run}, + db::{postgres_connection, redis_connection, setup_migrations, CampaignRemaining}, + Application, +}; +use slog::info; +use std::{convert::TryFrom, env, net::SocketAddr}; #[tokio::main] async fn main() -> Result<(), Box> { @@ -57,22 +54,12 @@ async fn main() -> Result<(), Box> { ) .get_matches(); - let environment = std::env::var("ENV").unwrap_or_else(|_| "development".into()); - let port = std::env::var("PORT") - .map(|s| s.parse::().expect("Invalid port(u16) was provided")) - .unwrap_or_else(|_| DEFAULT_PORT); + let env_config = sentry::application::Config::from_env()?; - let ip_addr = std::env::var("IP_ADDR") - .map(|s| { - s.parse::() - .expect("Invalid Ip address was provided") - }) - .unwrap_or_else(|_| DEFAULT_IP_ADDR); - - let socket_addr: SocketAddr = (ip_addr, port).into(); + let socket_addr: SocketAddr = (env_config.ip_addr, env_config.port).into(); let config_file = cli.value_of("config"); - let config = configuration(&environment, config_file).unwrap(); + let config = configuration(env_config.env, config_file).unwrap(); let adapter = match cli.value_of("adapter").unwrap() { "ethereum" => { @@ -108,13 +95,14 @@ async fn main() -> Result<(), Box> { _ => panic!("You can only use `ethereum` & `dummy` adapters!"), }; - let logger = logger(); - let url = env::var("REDIS_URL").unwrap_or_else(|_| String::from("redis://127.0.0.1:6379")); - let redis = redis_connection(url.as_str()).await?; + let logger = logger("sentry"); + let redis = redis_connection(env_config.redis_url).await?; info!(&logger, "Checking connection and applying migrations..."); // Check connection and setup migrations before setting up Postgres - setup_migrations(&environment).await; - let postgres = postgres_connection(42).await; + setup_migrations(env_config.env).await; + + // use the environmental variables to setup the Postgres connection + let postgres = postgres_connection(42, POSTGRES_CONFIG.clone()).await; let campaign_remaining = CampaignRemaining::new(redis.clone()); match adapter { @@ -150,36 +138,3 @@ async fn main() -> Result<(), Box> { Ok(()) } - -/// Starts the `hyper` `Server`. -async fn run(app: Application, socket_addr: SocketAddr) { - let logger = app.logger.clone(); - info!(&logger, "Listening on socket address: {}!", socket_addr); - - let make_service = make_service_fn(|_| { - let server = app.clone(); - async move { - Ok::<_, Error>(service_fn(move |req| { - let server = server.clone(); - async move { Ok::<_, Error>(server.handle_routing(req).await) } - })) - } - }); - - let server = Server::bind(&socket_addr).serve(make_service); - - if let Err(e) = server.await { - error!(&logger, "server error: {}", e; "main" => "run"); - } -} - -fn logger() -> Logger { - use primitives::util::logging::{Async, PrefixedCompactFormat, TermDecorator}; - use slog::{o, Drain}; - - let decorator = TermDecorator::new().build(); - let drain = PrefixedCompactFormat::new("sentry", decorator).fuse(); - let drain = Async::new(drain).build().fuse(); - - Logger::root(drain, o!()) -} diff --git a/sentry/src/middleware/auth.rs b/sentry/src/middleware/auth.rs index b65a7e8fd..99c17d707 100644 --- a/sentry/src/middleware/auth.rs +++ b/sentry/src/middleware/auth.rs @@ -129,11 +129,11 @@ fn get_request_ip(req: &Request) -> Option { mod test { use adapter::DummyAdapter; use hyper::Request; - use primitives::adapter::DummyAdapterOptions; - - use primitives::util::tests::prep_db::{AUTH, IDS}; - - use primitives::config::configuration; + use primitives::{ + adapter::DummyAdapterOptions, + config::DEVELOPMENT_CONFIG, + util::tests::prep_db::{AUTH, IDS}, + }; use deadpool::managed::Object; @@ -149,7 +149,7 @@ mod test { dummy_auth: IDS.clone(), dummy_auth_tokens: AUTH.clone(), }; - let config = configuration("development", None).expect("Dev config should be available"); + let config = DEVELOPMENT_CONFIG.clone(); (DummyAdapter::init(adapter_options, &config), connection) } diff --git a/sentry/src/routes/campaign.rs b/sentry/src/routes/campaign.rs index adb9de9c9..da8972e33 100644 --- a/sentry/src/routes/campaign.rs +++ b/sentry/src/routes/campaign.rs @@ -140,9 +140,10 @@ pub async fn create_campaign( let campaign = serde_json::from_slice::(&body) .map_err(|e| ResponseError::FailedValidation(e.to_string()))? - // create the actual `Campaign` with random `CampaignId` + // create the actual `Campaign` with a randomly generated `CampaignId` or the set `CampaignId` .into_campaign(); + // Validate the campaign as soon as a valid JSON was passed. campaign .validate(&app.config, &app.adapter.whoami()) .map_err(|err| ResponseError::FailedValidation(err.to_string()))?; @@ -957,7 +958,7 @@ mod test { let campaign: Campaign = { // erases the CampaignId for the CreateCampaign request - let mut create = CreateCampaign::from(dummy_campaign); + let mut create = CreateCampaign::from_campaign_erased(dummy_campaign, None); create.budget = UnifiedNum::from(500 * multiplier); // prepare for Campaign creation add_deposit_call(create.channel.id(), create.creator, create.channel.token); @@ -1030,7 +1031,8 @@ mod test { // we have 1000 left from our deposit, so we are using half of it let _second_campaign = { // erases the CampaignId for the CreateCampaign request - let mut create_second = CreateCampaign::from(DUMMY_CAMPAIGN.clone()); + let mut create_second = + CreateCampaign::from_campaign_erased(DUMMY_CAMPAIGN.clone(), None); create_second.budget = UnifiedNum::from(500 * multiplier); // prepare for Campaign creation @@ -1060,7 +1062,7 @@ mod test { // new campaign budget: 600 { // erases the CampaignId for the CreateCampaign request - let mut create = CreateCampaign::from(DUMMY_CAMPAIGN.clone()); + let mut create = CreateCampaign::from_campaign_erased(DUMMY_CAMPAIGN.clone(), None); create.budget = UnifiedNum::from(600 * multiplier); // prepare for Campaign creation @@ -1119,7 +1121,7 @@ mod test { // new campaign budget: 600 { // erases the CampaignId for the CreateCampaign request - let mut create = CreateCampaign::from(DUMMY_CAMPAIGN.clone()); + let mut create = CreateCampaign::from_campaign_erased(DUMMY_CAMPAIGN.clone(), None); create.budget = UnifiedNum::from(600 * multiplier); // prepare for Campaign creation diff --git a/test_harness/Cargo.toml b/test_harness/Cargo.toml new file mode 100644 index 000000000..be4d53775 --- /dev/null +++ b/test_harness/Cargo.toml @@ -0,0 +1,29 @@ +[package] +edition = "2018" +name = "test_harness" +version = "0.1.0" +authors = ["Ambire ", "Lachezar Lechev "] + +[dependencies] +primitives = { path = "../primitives", features = ["postgres"] } +adapter = { version = "0.1", path = "../adapter", features = ["test-util"] } +sentry = { version = "0.1", path = "../sentry", features = ["test-util"] } + +chrono = { version = "0.4", features = ["serde"] } + +# ethereum +web3 = { version = "0.17", features = ["http-tls", "signing"] } +once_cell = "^1.8" +reqwest = { version = "0.11", features = ["json"] } + +serde_json = { version = "1" } + +slog = { version = "^2.2.3", features = ["max_level_trace"] } +futures = "0.3" + +subprocess = "0.2" + +anyhow = { version = "1" } +tokio = { version = "1", features = ["rt-multi-thread", "macros", "time"] } +# probably needed for Relayer calls +# wiremock = "0.5" diff --git a/test_harness/Makefile.toml b/test_harness/Makefile.toml new file mode 100644 index 000000000..375d6d9ad --- /dev/null +++ b/test_harness/Makefile.toml @@ -0,0 +1,31 @@ +[tasks.dev-test-flow] +description = "Development testing flow will first format the code, and than run cargo build and test" +category = "Development" +dependencies = [ + "format-flow", + "format-toml-conditioned-flow", + "pre-build", + "build", + "post-build", + "local-test-flow", +] + +[tasks.test] +# run tests in release because of slow unlock time of Ethereum adapter (i.e. keystore decryption) +args = ["test", "--release"] + +[tasks.local-test-flow] +dependencies = ["services-up", "test-flow", "services-down"] + + +[tasks.services-up] +# `--renew-anon-volumes` will force the recreation of the services +# it's used primarily for `ganache-cli`. +# This forces the snapshot from previous unsuccessful test runs to get destroyed. +script = ''' +docker-compose -f ../docker-compose.harness.yml up --renew-anon-volumes -d ganache adex-redis adex-postgres \ +&& sleep 6 +''' + +[tasks.services-down] +script = "docker-compose -f ../docker-compose.harness.yml down" diff --git a/test_harness/src/deposits.rs b/test_harness/src/deposits.rs new file mode 100644 index 000000000..409e78a23 --- /dev/null +++ b/test_harness/src/deposits.rs @@ -0,0 +1,20 @@ +use primitives::{config::TokenInfo, Address, BigNum, Channel}; + +#[derive(Debug, Clone)] +pub struct Deposit { + pub channel: Channel, + pub token: TokenInfo, + pub address: Address, + /// In native token precision + pub outpace_amount: BigNum, + /// In native token precision + pub counterfactual_amount: BigNum, +} + +impl PartialEq> for Deposit { + fn eq(&self, other: &primitives::Deposit) -> bool { + let total = &self.outpace_amount + &self.counterfactual_amount; + + self.counterfactual_amount == other.still_on_create2 && total == other.total + } +} diff --git a/test_harness/src/lib.rs b/test_harness/src/lib.rs new file mode 100644 index 000000000..af12e22ec --- /dev/null +++ b/test_harness/src/lib.rs @@ -0,0 +1,813 @@ +use std::{ + collections::HashMap, + net::{IpAddr, Ipv4Addr}, +}; + +use adapter::ethereum::{ + get_counterfactual_address, + test_util::{ + deploy_outpace_contract, deploy_sweeper_contract, deploy_token_contract, mock_set_balance, + outpace_deposit, GANACHE_URL, MOCK_TOKEN_ABI, + }, + OUTPACE_ABI, SWEEPER_ABI, +}; +use deposits::Deposit; +use once_cell::sync::Lazy; +use primitives::{adapter::KeystoreOptions, config::TokenInfo, Address, Config}; +use web3::{contract::Contract, transports::Http, types::H160, Web3}; + +pub mod deposits; + +pub static GANACHE_CONFIG: Lazy = Lazy::new(|| { + Config::try_toml(include_str!("../../docs/config/ganache.toml")) + .expect("Failed to parse ganache.toml config file") +}); + +/// ganache-cli setup with deployed contracts using the snapshot directory +pub static SNAPSHOT_CONTRACTS: Lazy = Lazy::new(|| { + use primitives::BigNum; + use std::num::NonZeroU8; + + let web3 = Web3::new(Http::new(GANACHE_URL).expect("failed to init transport")); + + let token_address = "0x9db7bff788522dbe8fa2e8cbd568a58c471ccd5e" + .parse::
() + .unwrap(); + let token = ( + // copied from deploy_token_contract + TokenInfo { + min_token_units_for_deposit: BigNum::from(10_u64.pow(18)), + precision: NonZeroU8::new(18).expect("should create NonZeroU8"), + // multiplier = 10^14 - 10^18 (token precision) = 10^-4 + // min_validator_fee = 1' * 10^-4 = 0.000_1 + min_validator_fee: BigNum::from(100_000_000_000_000), + }, + token_address, + Contract::from_json(web3.eth(), H160(token_address.to_bytes()), &MOCK_TOKEN_ABI).unwrap(), + ); + + let sweeper_address = "0xdd41b0069256a28972458199a3c9cf036384c156" + .parse::
() + .unwrap(); + + let sweeper = ( + sweeper_address, + Contract::from_json(web3.eth(), H160(sweeper_address.to_bytes()), &SWEEPER_ABI).unwrap(), + ); + + let outpace_address = "0xcb097e455b7159f902e2eb45562fc397ae6b0f3d" + .parse::
() + .unwrap(); + + let outpace = ( + outpace_address, + Contract::from_json(web3.eth(), H160(outpace_address.to_bytes()), &OUTPACE_ABI).unwrap(), + ); + + Contracts { + token, + sweeper, + outpace, + } +}); + +#[derive(Debug, Clone)] +pub struct TestValidator { + pub address: Address, + pub keystore: KeystoreOptions, + pub sentry_config: sentry::application::Config, + /// Prefix for the loggers + pub logger_prefix: String, + /// Postgres DB name + /// The rest of the Postgres values are taken from env. variables + pub db_name: String, +} + +pub static VALIDATORS: Lazy> = Lazy::new(|| { + use adapter::ethereum::test_util::GANACHE_KEYSTORES; + use primitives::config::Environment; + + vec![ + ( + "leader", + TestValidator { + address: GANACHE_KEYSTORES["leader"].0, + keystore: GANACHE_KEYSTORES["leader"].1.clone(), + sentry_config: sentry::application::Config { + env: Environment::Development, + port: 8005, + ip_addr: IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), + redis_url: "redis://127.0.0.1:6379/1".parse().unwrap(), + }, + logger_prefix: "sentry-leader".into(), + db_name: "harness_leader".into(), + }, + ), + ( + "follower", + TestValidator { + address: GANACHE_KEYSTORES["follower"].0, + keystore: GANACHE_KEYSTORES["follower"].1.clone(), + sentry_config: sentry::application::Config { + env: Environment::Development, + port: 8006, + ip_addr: IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), + redis_url: "redis://127.0.0.1:6379/2".parse().unwrap(), + }, + logger_prefix: "sentry-follower".into(), + db_name: "harness_follower".into(), + }, + ), + ] + .into_iter() + .collect() +}); + +pub struct Setup { + pub web3: Web3, +} + +#[derive(Debug, Clone)] +pub struct Contracts { + pub token: (TokenInfo, Address, Contract), + pub sweeper: (Address, Contract), + pub outpace: (Address, Contract), +} + +impl Setup { + pub async fn deploy_contracts(&self) -> Contracts { + // deploy contracts + // TOKEN contract is with precision 18 (like DAI) + // set the minimum token units to 1 TOKEN + let token = deploy_token_contract(&self.web3, 10_u64.pow(18)) + .await + .expect("Correct parameters are passed to the Token constructor."); + + let sweeper = deploy_sweeper_contract(&self.web3) + .await + .expect("Correct parameters are passed to the Sweeper constructor."); + + let outpace = deploy_outpace_contract(&self.web3) + .await + .expect("Correct parameters are passed to the OUTPACE constructor."); + + Contracts { + token, + sweeper, + outpace, + } + } + + pub async fn deposit(&self, contracts: &Contracts, deposit: &Deposit) { + let counterfactual_address = get_counterfactual_address( + contracts.sweeper.0, + &deposit.channel, + contracts.outpace.0, + deposit.address, + ); + + // OUTPACE regular deposit + // first set a balance of tokens to be deposited + mock_set_balance( + &contracts.token.2, + deposit.address.to_bytes(), + deposit.address.to_bytes(), + &deposit.outpace_amount, + ) + .await + .expect("Failed to set balance"); + // call the OUTPACE deposit + outpace_deposit( + &contracts.outpace.1, + &deposit.channel, + deposit.address.to_bytes(), + &deposit.outpace_amount, + ) + .await + .expect("Should deposit with OUTPACE"); + + // Counterfactual address deposit + mock_set_balance( + &contracts.token.2, + deposit.address.to_bytes(), + counterfactual_address.to_bytes(), + &deposit.counterfactual_amount, + ) + .await + .expect("Failed to set balance"); + } +} + +#[cfg(test)] +mod tests { + use crate::run::run_sentry_app; + + use super::*; + use adapter::ethereum::{ + test_util::{GANACHE_ADDRESSES, GANACHE_KEYSTORES, GANACHE_URL}, + EthereumAdapter, + }; + use primitives::{ + adapter::Adapter, sentry::campaign_create::CreateCampaign, util::ApiUrl, BigNum, Campaign, + Channel, ChannelId, UnifiedNum, + }; + use reqwest::{Client, StatusCode}; + + #[tokio::test] + #[ignore = "We use a snapshot, however, we have left this test for convenience"] + async fn deploy_contracts() { + let web3 = Web3::new(Http::new(&GANACHE_URL).expect("failed to init transport")); + let setup = Setup { web3 }; + // deploy contracts + let _contracts = setup.deploy_contracts().await; + } + + static CAMPAIGN_1: Lazy = Lazy::new(|| { + use chrono::{TimeZone, Utc}; + use primitives::{ + campaign::{Active, Pricing, PricingBounds, Validators}, + targeting::Rules, + validator::ValidatorDesc, + EventSubmission, + }; + + let channel = Channel { + leader: VALIDATORS["leader"].address.into(), + follower: VALIDATORS["follower"].address.into(), + guardian: GANACHE_ADDRESSES["guardian"].into(), + token: SNAPSHOT_CONTRACTS.token.1, + nonce: 0_u64.into(), + }; + + let leader_desc = ValidatorDesc { + id: VALIDATORS["leader"].address.into(), + url: "http://localhost:8005".to_string(), + // fee per 1000 (pro mille) = 0.03000000 (UnifiedNum) + fee: 3_000_000.into(), + fee_addr: None, + }; + + let follower_desc = ValidatorDesc { + id: VALIDATORS["follower"].address.into(), + url: "http://localhost:8006".to_string(), + // fee per 1000 (pro mille) = 0.02000000 (UnifiedNum) + fee: 2_000_000.into(), + fee_addr: None, + }; + + let validators = Validators::new((leader_desc, follower_desc)); + + Campaign { + id: "0x936da01f9abd4d9d80c702af85c822a8" + .parse() + .expect("Should parse"), + channel, + creator: GANACHE_ADDRESSES["advertiser"], + // 20.00000000 + budget: UnifiedNum::from(200_000_000), + validators, + title: Some("Dummy Campaign".to_string()), + pricing_bounds: Some(PricingBounds { + impression: Some(Pricing { + // 0.00000100 + // Per 1000 = 0.00100000 + min: 100.into(), + // 0.00000200 + // Per 1000 = 0.00200000 + max: 200.into(), + }), + click: Some(Pricing { + // 0.00000300 + // Per 1000 = 0.00300000 + min: 300.into(), + // 0.00000500 + // Per 1000 = 0.00500000 + max: 500.into(), + }), + }), + event_submission: Some(EventSubmission { allow: vec![] }), + ad_units: vec![], + targeting_rules: Rules::new(), + created: Utc.ymd(2021, 2, 1).and_hms(7, 0, 0), + active: Active { + to: Utc.ymd(2099, 1, 30).and_hms(0, 0, 0), + from: None, + }, + } + }); + + /// This Campaign's Channel has switched leader & follower compared to [`CAMPAIGN_1`] + /// + /// `Channel.leader = VALIDATOR["follower"].address` + /// `Channel.follower = VALIDATOR["leader"],address` + /// See [`VALIDATORS`] for more details. + static CAMPAIGN_2: Lazy = Lazy::new(|| { + use chrono::{TimeZone, Utc}; + use primitives::{ + campaign::{Active, Pricing, PricingBounds, Validators}, + targeting::Rules, + validator::ValidatorDesc, + EventSubmission, + }; + + let channel = Channel { + leader: VALIDATORS["follower"].address.into(), + follower: VALIDATORS["leader"].address.into(), + guardian: GANACHE_ADDRESSES["guardian2"].into(), + token: SNAPSHOT_CONTRACTS.token.1, + nonce: 0_u64.into(), + }; + + // Uses the VALIDATORS["follower"] as the Leader for this Channel + // switches the URL as well + let leader_desc = ValidatorDesc { + id: VALIDATORS["follower"].address.into(), + url: "http://localhost:8006".to_string(), + // fee per 1000 (pro mille) = 0.10000000 (UnifiedNum) + fee: 10_000_000.into(), + fee_addr: None, + }; + + // Uses the VALIDATORS["leader"] as the Follower for this Channel + // switches the URL as well + let follower_desc = ValidatorDesc { + id: VALIDATORS["leader"].address.into(), + url: "http://localhost:8005".to_string(), + // fee per 1000 (pro mille) = 0.05000000 (UnifiedNum) + fee: 5_000_000.into(), + fee_addr: None, + }; + + let validators = Validators::new((leader_desc, follower_desc)); + + Campaign { + id: "0x127b98248f4e4b73af409d10f62daeaa" + .parse() + .expect("Should parse"), + channel, + creator: GANACHE_ADDRESSES["advertiser"], + // 20.00000000 + budget: UnifiedNum::from(20_00_000_000), + validators, + title: Some("Dummy Campaign".to_string()), + pricing_bounds: Some(PricingBounds { + impression: Some(Pricing { + // 0.00000100 + // Per 1000 = 0.00100000 + min: 100.into(), + // 0.00000200 + // Per 1000 = 0.00200000 + max: 200.into(), + }), + click: Some(Pricing { + // 0.00000300 + // Per 1000 = 0.00300000 + min: 300.into(), + // 0.00000500 + // Per 1000 = 0.00500000 + max: 500.into(), + }), + }), + event_submission: Some(EventSubmission { allow: vec![] }), + ad_units: vec![], + targeting_rules: Rules::new(), + created: Utc.ymd(2021, 2, 1).and_hms(7, 0, 0), + active: Active { + to: Utc.ymd(2099, 1, 30).and_hms(0, 0, 0), + from: None, + }, + } + }); + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn run_full_test() { + let web3 = Web3::new(Http::new(&GANACHE_URL).expect("failed to init transport")); + let setup = Setup { web3 }; + // Use snapshot contracts + let contracts = SNAPSHOT_CONTRACTS.clone(); + + let leader = VALIDATORS["leader"].clone(); + let follower = VALIDATORS["follower"].clone(); + + // let channel_1 = Channel { + // leader: leader.address.into(), + // follower: follower.address.into(), + // guardian: GANACHE_ADDRESSES["guardian"].into(), + // token: contracts.token.1, + // nonce: 0_u64.into(), + // }; + + // // switch the roles of the 2 validators & use a new guardian + // let channel_2 = Channel { + // leader: follower.address.into(), + // follower: leader.address.into(), + // guardian: GANACHE_ADDRESSES["guardian2"].into(), + // token: contracts.token.1, + // nonce: 1_u64.into(), + // }; + let token_precision = contracts.token.0.precision.get(); + + // We use the Advertiser's `EthereumAdapter::get_auth` for authentication! + let mut advertiser_adapter = + EthereumAdapter::init(GANACHE_KEYSTORES["advertiser"].1.clone(), &GANACHE_CONFIG) + .expect("Should initialize creator adapter"); + advertiser_adapter + .unlock() + .expect("Should unlock advertiser's Ethereum Adapter"); + let advertiser_adapter = advertiser_adapter; + + // setup Sentry & returns Adapter + let leader_adapter = setup_sentry(leader).await; + let follower_adapter = setup_sentry(follower).await; + + // Advertiser deposits + // + // Channel 1: + // - Outpace: 20 TOKENs + // - Counterfactual: 10 TOKENs + // + // Channel 2: + // - Outpace: 30 TOKENs + // - Counterfactual: 20 TOKENs + { + let advertiser_deposits = [ + Deposit { + channel: CAMPAIGN_1.channel, + token: contracts.token.0.clone(), + address: advertiser_adapter.whoami().to_address(), + outpace_amount: BigNum::with_precision(20, token_precision), + counterfactual_amount: BigNum::with_precision(10, token_precision), + }, + Deposit { + channel: CAMPAIGN_2.channel, + token: contracts.token.0.clone(), + address: advertiser_adapter.whoami().to_address(), + outpace_amount: BigNum::with_precision(30, token_precision), + counterfactual_amount: BigNum::with_precision(20, token_precision), + }, + ]; + // 1st deposit + { + setup.deposit(&contracts, &advertiser_deposits[0]).await; + + // make sure we have the expected deposit returned from EthereumAdapter + let eth_deposit = leader_adapter + .get_deposit( + &CAMPAIGN_1.channel, + &advertiser_adapter.whoami().to_address(), + ) + .await + .expect("Should get deposit for advertiser"); + + assert_eq!(advertiser_deposits[0], eth_deposit); + } + + // 2nd deposit + { + setup.deposit(&contracts, &advertiser_deposits[1]).await; + + // make sure we have the expected deposit returned from EthereumAdapter + let eth_deposit = leader_adapter + .get_deposit( + &CAMPAIGN_2.channel, + &advertiser_adapter.whoami().to_address(), + ) + .await + .expect("Should get deposit for advertiser"); + + assert_eq!(advertiser_deposits[1], eth_deposit); + } + } + + let api_client = reqwest::Client::new(); + let leader_url = CAMPAIGN_1.validators[0].try_api_url().expect("Valid url"); + let follower_url = CAMPAIGN_1.validators[1].try_api_url().expect("Valid url"); + + // No Channel 1 - 404 + // GET /v5/channel/{}/spender/all + { + let leader_auth = advertiser_adapter + .get_auth(&leader_adapter.whoami()) + .expect("Get authentication"); + + let leader_response = get_spender_all_page_0( + &api_client, + &leader_url, + &leader_auth, + CAMPAIGN_1.channel.id(), + ) + .await + .expect("Should return Response"); + + assert_eq!(StatusCode::NOT_FOUND, leader_response.status()); + } + + // Create Campaign 1 w/ Channel 1 using Advertiser + // Response: 400 - not enough deposit + // Channel 1 - Is created, even though campaign creation failed. + // POST /v5/campaign + { + let leader_auth = advertiser_adapter + .get_auth(&leader_adapter.whoami()) + .expect("Get authentication"); + + let mut no_budget_campaign = CreateCampaign::from_campaign(CAMPAIGN_1.clone()); + // Deposit of Advertiser for Channel 2: 20 (outpace) + 10 (create2) + // Campaign Budget: 40 TOKENs + no_budget_campaign.budget = UnifiedNum::from(4_000_000_000); + + let no_budget_response = + create_campaign(&api_client, &leader_url, &leader_auth, &no_budget_campaign) + .await + .expect("Should return Response"); + let status = no_budget_response.status(); + let response = no_budget_response + .json::() + .await + .expect("Deserialization"); + + assert_eq!(StatusCode::BAD_REQUEST, status); + let expected_error = serde_json::json!({ + "message": "Not enough deposit left for the new campaign's budget" + }); + + assert_eq!(expected_error, response); + } + + // Channel 1 - 200 + // Exists from the previously failed create Campaign 1 request + // GET /v5/channel/{}/spender/all + { + let leader_auth = advertiser_adapter + .get_auth(&leader_adapter.whoami()) + .expect("Get authentication"); + + let leader_response = get_spender_all_page_0( + &api_client, + &leader_url, + &leader_auth, + CAMPAIGN_1.channel.id(), + ) + .await + .expect("Should return Response"); + + assert_eq!(StatusCode::OK, leader_response.status()); + } + + // Create Campaign 1 w/ Channel 1 using Advertiser + // In Leader & Follower sentries + // Response: 200 Ok + { + let create_campaign_1 = CreateCampaign::from_campaign(CAMPAIGN_1.clone()); + { + let leader_token = advertiser_adapter + .get_auth(&leader_adapter.whoami()) + .expect("Get authentication"); + + let leader_response = + create_campaign(&api_client, &leader_url, &leader_token, &create_campaign_1) + .await + .expect("Should return Response"); + + assert_eq!(StatusCode::OK, leader_response.status()); + } + + { + let follower_token = advertiser_adapter + .get_auth(&follower_adapter.whoami()) + .expect("Get authentication"); + + let follower_response = create_campaign( + &api_client, + &follower_url, + &follower_token, + &create_campaign_1, + ) + .await + .expect("Should return Response"); + + assert_eq!(StatusCode::OK, follower_response.status()); + } + } + + // Create Campaign 2 w/ Channel 2 using Advertiser + // In Leader & Follower sentries + // Response: 200 Ok + // POST /v5/campaign + { + let create_campaign_2 = CreateCampaign::from_campaign(CAMPAIGN_2.clone()); + + { + let leader_token = advertiser_adapter + .get_auth(&leader_adapter.whoami()) + .expect("Get authentication"); + + let leader_response = + create_campaign(&api_client, &leader_url, &leader_token, &create_campaign_2) + .await + .expect("Should return Response"); + let status = leader_response.status(); + + assert_eq!(StatusCode::OK, status); + } + + { + let follower_token = advertiser_adapter + .get_auth(&follower_adapter.whoami()) + .expect("Get authentication"); + + let follower_response = create_campaign( + &api_client, + &follower_url, + &follower_token, + &create_campaign_2, + ) + .await + .expect("Should return Response"); + + assert_eq!(StatusCode::OK, follower_response.status()); + } + } + + // setup worker + + // run worker single-tick + } + + async fn setup_sentry(validator: TestValidator) -> EthereumAdapter { + let mut adapter = EthereumAdapter::init(validator.keystore, &GANACHE_CONFIG) + .expect("EthereumAdapter::init"); + + adapter.unlock().expect("Unlock successfully adapter"); + + run_sentry_app( + adapter.clone(), + &validator.logger_prefix, + validator.sentry_config, + &validator.db_name, + ) + .await + .expect("To run Sentry API server"); + + adapter + } + + async fn get_spender_all_page_0( + api_client: &Client, + url: &ApiUrl, + token: &str, + channel: ChannelId, + ) -> anyhow::Result { + let endpoint_url = url + .join(&format!("v5/channel/{}/spender/all", channel)) + .expect("valid endpoint"); + + Ok(api_client + .get(endpoint_url) + .bearer_auth(&token) + .send() + .await?) + } + + async fn create_campaign( + api_client: &Client, + url: &ApiUrl, + token: &str, + create_campaign: &CreateCampaign, + ) -> anyhow::Result { + let endpoint_url = url.join("v5/campaign").expect("valid endpoint"); + + Ok(api_client + .post(endpoint_url) + .json(create_campaign) + .bearer_auth(token) + .send() + .await?) + } +} +pub mod run { + use std::{env::current_dir, net::SocketAddr, path::PathBuf}; + + use adapter::EthereumAdapter; + use primitives::{ + postgres::{POSTGRES_HOST, POSTGRES_PASSWORD, POSTGRES_PORT, POSTGRES_USER}, + ToETHChecksum, ValidatorId, + }; + use sentry::{ + application::{logger, run}, + db::{ + postgres_connection, redis_connection, redis_pool::Manager, + tests_postgres::setup_test_migrations, CampaignRemaining, + }, + Application, + }; + use slog::info; + use subprocess::{Popen, PopenConfig, Redirection}; + + use crate::GANACHE_CONFIG; + + pub async fn run_sentry_app( + adapter: EthereumAdapter, + logger_prefix: &str, + app_config: sentry::application::Config, + db_name: &str, + ) -> anyhow::Result<()> { + let socket_addr = SocketAddr::new(app_config.ip_addr, app_config.port); + + let postgres_config = { + let mut config = sentry::db::PostgresConfig::new(); + + config + .user(POSTGRES_USER.as_str()) + .password(POSTGRES_PASSWORD.as_str()) + .host(POSTGRES_HOST.as_str()) + .port(*POSTGRES_PORT) + .dbname(db_name); + + config + }; + + let postgres = postgres_connection(42, postgres_config).await; + let mut redis = redis_connection(app_config.redis_url).await?; + + Manager::flush_db(&mut redis).await.expect("Should flush redis database"); + + let campaign_remaining = CampaignRemaining::new(redis.clone()); + + setup_test_migrations(postgres.clone()) + .await + .expect("Should run migrations"); + + let app = Application::new( + adapter, + GANACHE_CONFIG.clone(), + logger(logger_prefix), + redis, + postgres, + campaign_remaining, + ); + + info!(&app.logger, "Spawn sentry Hyper server"); + tokio::spawn(run(app, socket_addr)); + + Ok(()) + } + /// This helper function generates the correct file path to a project file from the current one. + /// + /// The `file_path` starts from the Cargo workspace directory. + fn project_file_path(file_path: &str) -> PathBuf { + let full_path = current_dir().unwrap(); + let project_path = full_path.parent().unwrap().to_path_buf(); + + project_path.join(file_path) + } + + /// ```bash + /// POSTGRES_DB=sentry_leader PORT=8005 KEYSTORE_PWD=address1 \ + /// cargo run -p sentry -- --adapter ethereum --keystoreFile ./adapter/test/resources/0x5a04A8fB90242fB7E1db7d1F51e268A03b7f93A5_keystore.json \ + /// ./docs/config/ganache.toml + /// ``` + /// + /// The identity is used to get the correct Keystore file + /// While the password is passed to `sentry` with environmental variable + pub fn run_sentry(keystore_password: &str, identity: ValidatorId) -> anyhow::Result { + let keystore_file_name = format!( + "adapter/test/resources/{}_keystore.json", + identity.to_checksum() + ); + let keystore_path = project_file_path(&keystore_file_name); + let ganache_config_path = project_file_path("docs/config/ganache.toml"); + + let sentry_leader = Popen::create( + &[ + "cargo", + "run", + "-p", + "sentry", + "--", + "--adapter", + "ethereum", + "--keystoreFile", + &keystore_path.to_string_lossy(), + &ganache_config_path.to_string_lossy(), + ], + PopenConfig { + stdout: Redirection::Pipe, + env: Some(vec![ + ("PORT".parse().unwrap(), "8005".parse().unwrap()), + ( + "POSTGRES_DB".parse().unwrap(), + "sentry_leader".parse().unwrap(), + ), + ( + "KEYSTORE_PWD".parse().unwrap(), + keystore_password.parse().unwrap(), + ), + ]), + ..Default::default() + }, + )?; + + Ok(sentry_leader) + } +} diff --git a/test_harness/src/main.rs b/test_harness/src/main.rs new file mode 100644 index 000000000..7f755fb76 --- /dev/null +++ b/test_harness/src/main.rs @@ -0,0 +1,2 @@ +#[tokio::main] +async fn main() {} diff --git a/validator_worker/Cargo.toml b/validator_worker/Cargo.toml index 82424e682..e8400fe62 100644 --- a/validator_worker/Cargo.toml +++ b/validator_worker/Cargo.toml @@ -31,7 +31,6 @@ tokio = { version = "1", features = ["time", "rt-multi-thread"] } # API client reqwest = { version = "0.11", features = ["json"] } # Other -lazy_static = "^1.4" thiserror = "^1.0" # (De)Serialization serde = { version = "^1.0", features = ["derive"] } diff --git a/validator_worker/src/lib.rs b/validator_worker/src/lib.rs index e5cb4682f..5fc10d7f2 100644 --- a/validator_worker/src/lib.rs +++ b/validator_worker/src/lib.rs @@ -18,6 +18,7 @@ pub mod follower; pub mod heartbeat; pub mod leader; pub mod sentry_interface; +pub mod worker; pub mod core { pub mod follower_rules; diff --git a/validator_worker/src/main.rs b/validator_worker/src/main.rs index ca29f5596..8dbaf72b4 100644 --- a/validator_worker/src/main.rs +++ b/validator_worker/src/main.rs @@ -1,38 +1,21 @@ #![deny(rust_2018_idioms)] #![deny(clippy::all)] -use std::{convert::TryFrom, error::Error, time::Duration}; +use std::{convert::TryFrom, error::Error}; use clap::{crate_version, App, Arg}; -use futures::{ - future::{join, join_all}, - TryFutureExt, -}; -use tokio::{runtime::Runtime, time::sleep}; use adapter::{AdapterTypes, DummyAdapter, EthereumAdapter}; use primitives::{ - adapter::{Adapter, DummyAdapterOptions, KeystoreOptions}, - config::{configuration, Config}, + adapter::{DummyAdapterOptions, KeystoreOptions}, + config::{configuration, Environment}, util::{ + logging::new_logger, tests::prep_db::{AUTH, IDS}, - ApiUrl, }, ValidatorId, }; -use slog::{error, info, Logger}; -use std::fmt::Debug; -use validator_worker::{ - channel::{channel_tick, collect_channels}, - SentryApi, -}; - -#[derive(Debug, Clone)] -struct Args { - sentry_url: ApiUrl, - config: Config, - adapter: A, -} +use validator_worker::worker::run; fn main() -> Result<(), Box> { let cli = App::new("Validator worker") @@ -84,9 +67,13 @@ fn main() -> Result<(), Box> { ) .get_matches(); - let environment = std::env::var("ENV").unwrap_or_else(|_| "development".into()); + let environment: Environment = serde_json::from_value(serde_json::Value::String( + std::env::var("ENV").expect("Valid environment variable"), + )) + .expect("Valid Environment - development or production"); + let config_file = cli.value_of("config"); - let config = configuration(&environment, config_file).expect("failed to parse configuration"); + let config = configuration(environment, config_file).expect("failed to parse configuration"); let sentry_url = cli .value_of("sentryUrl") .expect("sentry url missing") @@ -122,7 +109,7 @@ fn main() -> Result<(), Box> { _ => panic!("We don't have any other adapters implemented yet!"), }; - let logger = logger(); + let logger = new_logger("validator_worker"); match adapter { AdapterTypes::EthereumAdapter(ethadapter) => { @@ -133,98 +120,3 @@ fn main() -> Result<(), Box> { } } } - -fn run( - is_single_tick: bool, - sentry_url: ApiUrl, - config: &Config, - mut adapter: A, - logger: &Logger, -) -> Result<(), Box> { - // unlock adapter - adapter.unlock()?; - - let args = Args { - sentry_url, - config: config.to_owned(), - adapter, - }; - - // Create the runtime - let rt = Runtime::new()?; - - if is_single_tick { - rt.block_on(all_channels_tick(args, logger)); - } else { - rt.block_on(infinite(args, logger)); - } - - Ok(()) -} - -async fn infinite(args: Args, logger: &Logger) { - loop { - let arg = args.clone(); - let wait_time_future = sleep(Duration::from_millis(arg.config.wait_time as u64)); - - let _result = join(all_channels_tick(arg, logger), wait_time_future).await; - } -} - -async fn all_channels_tick(args: Args, logger: &Logger) { - let (channels, validators) = match collect_channels( - &args.adapter, - &args.sentry_url, - &args.config, - logger, - ) - .await - { - Ok(res) => res, - Err(err) => { - error!(logger, "Error collecting all channels for tick"; "collect_channels" => ?err, "main" => "all_channels_tick"); - return; - } - }; - let channels_size = channels.len(); - - // initialize SentryApi once we have all the Campaign Validators we need to propagate messages to - let sentry = match SentryApi::init( - args.adapter.clone(), - logger.clone(), - args.config.clone(), - validators.clone(), - ) { - Ok(sentry) => sentry, - Err(err) => { - error!(logger, "Failed to initialize SentryApi for all channels"; "SentryApi::init()" => ?err, "main" => "all_channels_tick"); - return; - } - }; - - let tick_results = join_all(channels.into_iter().map(|channel| { - channel_tick(&sentry, &args.config, channel).map_err(move |err| (channel, err)) - })) - .await; - - for (channel, channel_err) in tick_results.into_iter().filter_map(Result::err) { - error!(logger, "Error processing Channel"; "channel" => ?channel, "error" => ?channel_err, "main" => "all_channels_tick"); - } - - info!(logger, "Processed {} channels", channels_size); - - if channels_size >= args.config.max_channels as usize { - error!(logger, "WARNING: channel limit cfg.MAX_CHANNELS={} reached", &args.config.max_channels; "main" => "all_channels_tick"); - } -} - -fn logger() -> Logger { - use primitives::util::logging::{Async, PrefixedCompactFormat, TermDecorator}; - use slog::{o, Drain}; - - let decorator = TermDecorator::new().build(); - let drain = PrefixedCompactFormat::new("validator_worker", decorator).fuse(); - let drain = Async::new(drain).build().fuse(); - - Logger::root(drain, o!()) -} diff --git a/validator_worker/src/sentry_interface.rs b/validator_worker/src/sentry_interface.rs index e36591b19..db3f75f57 100644 --- a/validator_worker/src/sentry_interface.rs +++ b/validator_worker/src/sentry_interface.rs @@ -139,7 +139,7 @@ impl SentryApi { .await } - /// Get's the last approved state and requesting a [`Heartbeat`], see [`LastApprovedResponse`] + /// Get's the last approved state and requesting a [`primitives::validator::Heartbeat`], see [`LastApprovedResponse`] pub async fn get_last_approved( &self, channel: ChannelId, @@ -319,6 +319,7 @@ pub mod channels { .await } } + pub mod campaigns { use chrono::Utc; use futures::future::try_join_all; diff --git a/validator_worker/src/worker.rs b/validator_worker/src/worker.rs new file mode 100644 index 000000000..167eab9cf --- /dev/null +++ b/validator_worker/src/worker.rs @@ -0,0 +1,104 @@ +use crate::{ + channel::{channel_tick, collect_channels}, + SentryApi, +}; +use primitives::{adapter::Adapter, util::ApiUrl, Config}; +use slog::{error, info, Logger}; +use std::{error::Error, time::Duration}; + +use futures::{ + future::{join, join_all}, + TryFutureExt, +}; +use tokio::{runtime::Runtime, time::sleep}; + +#[derive(Debug, Clone)] +pub struct Args { + sentry_url: ApiUrl, + config: Config, + adapter: A, +} + +pub fn run( + is_single_tick: bool, + sentry_url: ApiUrl, + config: &Config, + mut adapter: A, + logger: &Logger, +) -> Result<(), Box> { + // unlock adapter + adapter.unlock()?; + + let args = Args { + sentry_url, + config: config.to_owned(), + adapter, + }; + + // Create the runtime + let rt = Runtime::new()?; + + if is_single_tick { + rt.block_on(all_channels_tick(args, logger)); + } else { + rt.block_on(infinite(args, logger)); + } + + Ok(()) +} + +pub async fn infinite(args: Args, logger: &Logger) { + loop { + let arg = args.clone(); + let wait_time_future = sleep(Duration::from_millis(arg.config.wait_time as u64)); + + let _result = join(all_channels_tick(arg, logger), wait_time_future).await; + } +} + +pub async fn all_channels_tick(args: Args, logger: &Logger) { + let (channels, validators) = match collect_channels( + &args.adapter, + &args.sentry_url, + &args.config, + logger, + ) + .await + { + Ok(res) => res, + Err(err) => { + error!(logger, "Error collecting all channels for tick"; "collect_channels" => ?err, "main" => "all_channels_tick"); + return; + } + }; + let channels_size = channels.len(); + + // initialize SentryApi once we have all the Campaign Validators we need to propagate messages to + let sentry = match SentryApi::init( + args.adapter.clone(), + logger.clone(), + args.config.clone(), + validators.clone(), + ) { + Ok(sentry) => sentry, + Err(err) => { + error!(logger, "Failed to initialize SentryApi for all channels"; "SentryApi::init()" => ?err, "main" => "all_channels_tick"); + return; + } + }; + + let tick_results = join_all(channels.into_iter().map(|channel| { + channel_tick(&sentry, &args.config, channel).map_err(move |err| (channel, err)) + })) + .await; + + for (channel, channel_err) in tick_results.into_iter().filter_map(Result::err) { + error!(logger, "Error processing Channel"; "channel" => ?channel, "error" => ?channel_err, "main" => "all_channels_tick"); + } + + info!(logger, "Processed {} channels", channels_size); + + if channels_size >= args.config.max_channels as usize { + error!(logger, "WARNING: channel limit cfg.MAX_CHANNELS={} reached", &args.config.max_channels; "main" => "all_channels_tick"); + } +}