Merge branch 'breaking' into vb-smart-contracts-v8

core/bin/data_restore/src/database_storage_interactor.rs

@@ -120,7 +120,7 @@ impl<'a> DatabaseStorageInteractor<'a> {
         transaction
             .chain()
             .block_schema()
-            .save_block(block)
+            .save_full_block(block)
             .await
             .expect("Unable to save block");
 

core/bin/server/src/main.rs

@@ -6,7 +6,7 @@ use structopt::StructOpt;
 
 use serde::{Deserialize, Serialize};
 
-use zksync_api::fee_ticker::{run_ticker_task, TickerRequest};
+use zksync_api::fee_ticker::{run_updaters, FeeTicker, TickerInfo};
 use zksync_core::{genesis_init, run_core, wait_for_tasks};
 use zksync_eth_client::EthereumGateway;
 use zksync_forced_exit_requests::run_forced_exit_requests_actors;
@@ -42,6 +42,7 @@ pub enum Component {
     RpcWebSocketApi,
 
     // Core components
+    Fetchers,
     EthSender,
     Core,
     WitnessGenerator,
@@ -65,6 +66,7 @@ impl FromStr for Component {
             "witness-generator" => Ok(Component::WitnessGenerator),
             "forced-exit" => Ok(Component::ForcedExit),
             "prometheus" => Ok(Component::Prometheus),
+            "fetchers" => Ok(Component::Fetchers),
             "core" => Ok(Component::Core),
             "rejected-task-cleaner" => Ok(Component::RejectedTaskCleaner),
             other => Err(format!("{} is not a valid component name", other)),
@@ -88,6 +90,7 @@ impl Default for ComponentsToRun {
             Component::Prometheus,
             Component::Core,
             Component::RejectedTaskCleaner,
+            Component::Fetchers,
         ])
     }
 }
@@ -113,7 +116,7 @@ struct Opt {
     /// comma-separated list of components to launch
     #[structopt(
         long,
-        default_value = "rest-api,web3-api,rpc-api,rpc-websocket-api,eth-sender,witness-generator,forced-exit,prometheus,core,rejected-task-cleaner"
+        default_value = "rest-api,web3-api,rpc-api,rpc-websocket-api,eth-sender,witness-generator,forced-exit,prometheus,core,rejected-task-cleaner,fetchers"
     )]
     components: ComponentsToRun,
 }
@@ -159,6 +162,12 @@ async fn run_server(components: &ComponentsToRun) {
         ));
     }
 
+    if components.0.contains(&Component::Fetchers) {
+        // Run price fetchers
+        let mut price_tasks = run_price_updaters(connection_pool.clone());
+        tasks.append(&mut price_tasks);
+    }
+
     if components.0.iter().any(|c| {
         matches!(
             c,
@@ -178,10 +187,6 @@ async fn run_server(components: &ComponentsToRun) {
             tasks.push(task);
         }
 
-        // Run ticker
-        let (task, ticker_request_sender) = run_ticker(connection_pool.clone(), channel_size);
-        tasks.push(task);
-
         // Run signer
         let (sign_check_sender, sign_check_receiver) = mpsc::channel(channel_size);
         tasks.push(zksync_api::signature_checker::start_sign_checker(
@@ -193,12 +198,21 @@ async fn run_server(components: &ComponentsToRun) {
         let contracts_config = ContractsConfig::from_env();
         let common_config = CommonApiConfig::from_env();
         let chain_config = ChainConfig::from_env();
+        let ticker_info = Box::new(TickerInfo::new(connection_pool.clone()));
+        let fee_ticker_config = TickerConfig::from_env();
+
+        let ticker = FeeTicker::new_with_default_validator(
+            ticker_info,
+            fee_ticker_config,
+            chain_config.max_blocks_to_aggregate(),
+            connection_pool.clone(),
+        );
 
         if components.0.contains(&Component::RpcWebSocketApi) {
             tasks.push(zksync_api::api_server::rpc_subscriptions::start_ws_server(
                 connection_pool.clone(),
                 sign_check_sender.clone(),
-                ticker_request_sender.clone(),
+                ticker.clone(),
                 &common_config,
                 &JsonRpcConfig::from_env(),
                 chain_config.state_keeper.miniblock_iteration_interval(),
@@ -211,7 +225,7 @@ async fn run_server(components: &ComponentsToRun) {
             tasks.push(zksync_api::api_server::rpc_server::start_rpc_server(
                 connection_pool.clone(),
                 sign_check_sender.clone(),
-                ticker_request_sender.clone(),
+                ticker.clone(),
                 &JsonRpcConfig::from_env(),
                 &common_config,
                 private_config.url.clone(),
@@ -224,7 +238,7 @@ async fn run_server(components: &ComponentsToRun) {
                 connection_pool.clone(),
                 RestApiConfig::from_env().bind_addr(),
                 contracts_config.contract_addr,
-                ticker_request_sender,
+                ticker,
                 sign_check_sender,
                 private_config.url,
             );
@@ -330,21 +344,9 @@ pub fn run_eth_sender(connection_pool: ConnectionPool) -> JoinHandle<()> {
     zksync_eth_sender::run_eth_sender(connection_pool, eth_gateway, eth_sender_config)
 }
 
-pub fn run_ticker(
-    connection_pool: ConnectionPool,
-    channel_size: usize,
-) -> (JoinHandle<()>, mpsc::Sender<TickerRequest>) {
-    vlog::info!("Starting Ticker actors");
-    let (ticker_request_sender, ticker_request_receiver) = mpsc::channel(channel_size);
-    let chain_config = ChainConfig::from_env();
+pub fn run_price_updaters(connection_pool: ConnectionPool) -> Vec<JoinHandle<()>> {
     let ticker_config = TickerConfig::from_env();
-    let task = run_ticker_task(
-        connection_pool,
-        ticker_request_receiver,
-        &ticker_config,
-        chain_config.max_blocks_to_aggregate(),
-    );
-    (task, ticker_request_sender)
+    run_updaters(connection_pool, &ticker_config)
 }
 
 pub fn create_eth_gateway() -> EthereumGateway {

core/bin/zksync_api/src/api_server/rest/forced_exit_requests/error.rs

@@ -105,6 +105,7 @@ impl SumbitErrorCode {
             SubmitError::Internal(_) => Self::Internal,
             SubmitError::Other(_) => Self::Other,
             SubmitError::Toggle2FA(_) => Self::Other,
+            SubmitError::PriceError(_) => Self::Other,
         }
     }
 

core/bin/zksync_api/src/api_server/rest/mod.rs

@@ -8,10 +8,11 @@ use zksync_types::H160;
 use zksync_utils::panic_notify::{spawn_panic_handler, ThreadPanicNotify};
 
 use self::v01::api_decl::ApiV01;
-use crate::{fee_ticker::TickerRequest, signature_checker::VerifySignatureRequest};
+use crate::signature_checker::VerifySignatureRequest;
 
 use super::tx_sender::TxSender;
 
+use crate::fee_ticker::FeeTicker;
 use tokio::task::JoinHandle;
 use zksync_config::ZkSyncConfig;
 
@@ -22,7 +23,7 @@ pub mod v02;
 
 async fn start_server(
     api_v01: ApiV01,
-    fee_ticker: mpsc::Sender<TickerRequest>,
+    fee_ticker: FeeTicker,
     sign_verifier: mpsc::Sender<VerifySignatureRequest>,
     bind_to: SocketAddr,
 ) {
@@ -83,7 +84,7 @@ pub fn start_server_thread_detached(
     connection_pool: ConnectionPool,
     listen_addr: SocketAddr,
     contract_address: H160,
-    fee_ticker: mpsc::Sender<TickerRequest>,
+    fee_ticker: FeeTicker,
     sign_verifier: mpsc::Sender<VerifySignatureRequest>,
     private_url: String,
 ) -> JoinHandle<()> {

core/bin/zksync_api/src/api_server/rest/v01/api_impl.rs

@@ -522,13 +522,19 @@ impl ApiV01 {
             .block_schema()
             .get_last_saved_block()
             .await
-            .expect("Database error");
+            .map_err(|err| {
+                vlog::warn!("Internal Server Error: '{}';", err,);
+                InternalError::from_response(err, HttpResponse::InternalServerError().finish())
+            })?;
         let block = storage
             .chain()
             .block_schema()
             .get_block(block_number)
             .await
-            .expect("Database error")
+            .map_err(|err| {
+                vlog::warn!("Internal Server Error: '{}';", err);
+                InternalError::from_response(err, HttpResponse::InternalServerError().finish())
+            })?
             .expect("Should exist");
         let state_keeper_config = &self_.config.chain.state_keeper;
         let average_proof_generating_time = Duration::minutes(30);

core/bin/zksync_api/src/api_server/rest/v02/error.rs

@@ -174,6 +174,16 @@ impl ApiError for CoreApiError {
     }
 }
 
+impl ApiError for anyhow::Error {
+    fn error_type(&self) -> String {
+        "internalError".to_string()
+    }
+
+    fn code(&self) -> ErrorCode {
+        ErrorCode::Other
+    }
+}
+
 impl ApiError for SubmitError {
     fn error_type(&self) -> String {
         String::from("submitError")
@@ -191,6 +201,7 @@ impl ApiError for SubmitError {
             Self::Internal(_) => ErrorCode::InternalError,
             Self::Toggle2FA(_) => ErrorCode::Toggle2FAError,
             Self::Other(_) => ErrorCode::Other,
+            Self::PriceError(_) => ErrorCode::InternalError,
         }
     }
 }

core/bin/zksync_api/src/api_server/rest/v02/fee.rs

@@ -34,33 +34,35 @@ async fn get_tx_fee(
     data: web::Data<ApiFeeData>,
     Json(body): Json<TxFeeRequest>,
 ) -> ApiResult<ApiFee> {
-    let token_allowed = api_try!(TxSender::token_allowed_for_fees(
-        data.tx_sender.ticker_requests.clone(),
-        body.token_like.clone()
-    )
-    .await
-    .map_err(Error::from));
+    let token_allowed = api_try!(data
+        .tx_sender
+        .ticker
+        .token_allowed_for_fees(body.token_like.clone())
+        .await
+        .map_err(Error::from));
     if !token_allowed {
         return Error::from(SubmitError::InappropriateFeeToken).into();
     }
+    // TODO implement subsidies for v02 api ZKS-888
     data.tx_sender
-        .get_txs_fee_in_wei(body.tx_type.into(), body.address, body.token_like)
+        .ticker
+        .get_fee_from_ticker_in_wei(body.tx_type.into(), body.token_like, body.address)
         .await
+        .map(|fee| fee.normal_fee.into())
         .map_err(Error::from)
-        .map(ApiFee::from)
         .into()
 }
 
 async fn get_batch_fee(
     data: web::Data<ApiFeeData>,
     Json(body): Json<BatchFeeRequest>,
 ) -> ApiResult<ApiFee> {
-    let token_allowed = api_try!(TxSender::token_allowed_for_fees(
-        data.tx_sender.ticker_requests.clone(),
-        body.token_like.clone()
-    )
-    .await
-    .map_err(Error::from));
+    let token_allowed = api_try!(data
+        .tx_sender
+        .ticker
+        .token_allowed_for_fees(body.token_like.clone())
+        .await
+        .map_err(Error::from));
     if !token_allowed {
         return Error::from(SubmitError::InappropriateFeeToken).into();
     }
@@ -70,10 +72,11 @@ async fn get_batch_fee(
         .map(|tx| (tx.tx_type.into(), tx.address))
         .collect();
     data.tx_sender
-        .get_txs_batch_fee_in_wei(txs, body.token_like)
+        .ticker
+        .get_batch_from_ticker_in_wei(body.token_like, txs)
         .await
+        .map(|fee| fee.normal_fee.into())
         .map_err(Error::from)
-        .map(ApiFee::from)
         .into()
 }
 
@@ -95,12 +98,19 @@ mod tests {
         },
         SharedData,
     };
+    use crate::fee_ticker::validator::cache::TokenInMemoryCache;
+    use chrono::Utc;
+    use num::rational::Ratio;
     use num::BigUint;
+    use std::collections::HashMap;
     use zksync_api_types::v02::{
         fee::{ApiTxFeeTypes, TxInBatchFeeRequest},
         ApiVersion,
     };
-    use zksync_types::{tokens::TokenLike, Address, TokenId};
+    use zksync_types::{
+        tokens::{TokenLike, TokenMarketVolume},
+        Address, Token, TokenId, TokenKind,
+    };
 
     #[actix_rt::test]
     #[cfg_attr(
@@ -115,12 +125,38 @@ mod tests {
             api_version: ApiVersion::V02,
         };
 
+        let mut tokens = HashMap::new();
+        tokens.insert(
+            TokenLike::Id(TokenId(1)),
+            Token::new(TokenId(1), Default::default(), "", 18, TokenKind::ERC20),
+        );
+        tokens.insert(
+            TokenLike::Id(TokenId(2)),
+            Token::new(TokenId(2), Default::default(), "", 18, TokenKind::ERC20),
+        );
+        let mut market = HashMap::new();
+        market.insert(
+            TokenId(2),
+            TokenMarketVolume {
+                market_volume: Ratio::from_integer(BigUint::from(400u32)),
+                last_updated: Utc::now(),
+            },
+        );
+        let prices = vec![
+            (TokenLike::Id(TokenId(0)), 10_u64.into()),
+            (TokenLike::Id(TokenId(1)), 10_u64.into()),
+            (TokenLike::Id(TokenId(2)), 10000_u64.into()),
+        ];
+
+        let cache = TokenInMemoryCache::new()
+            .with_tokens(tokens)
+            .with_market(market);
         let (client, server) = cfg.start_server(
             move |cfg: &TestServerConfig| {
                 api_scope(TxSender::new(
                     cfg.pool.clone(),
                     dummy_sign_verifier(),
-                    dummy_fee_ticker(&[]),
+                    dummy_fee_ticker(&prices, Some(cache.clone())),
                     &cfg.config.api.common,
                     cfg.config.api.private.url.clone(),
                 ))
@@ -157,9 +193,9 @@ mod tests {
 
         let response = client.get_batch_fee(txs, allowed_token).await?;
         let api_batch_fee: ApiFee = deserialize_response_result(response)?;
-        assert_eq!(api_batch_fee.gas_fee, BigUint::from(3u32));
-        assert_eq!(api_batch_fee.zkp_fee, BigUint::from(3u32));
-        assert_eq!(api_batch_fee.total_fee, BigUint::from(6u32));
+        assert_eq!(api_batch_fee.gas_fee, BigUint::from(1u32));
+        assert_eq!(api_batch_fee.zkp_fee, BigUint::from(1u32));
+        assert_eq!(api_batch_fee.total_fee, BigUint::from(2u32));
 
         server.stop().await;
         Ok(())

core/bin/zksync_api/src/api_server/rest/v02/mod.rs

@@ -55,7 +55,7 @@ pub(crate) fn api_scope(tx_sender: TxSender, zk_config: &ZkSyncConfig) -> Scope
             zk_config,
             tx_sender.pool.clone(),
             tx_sender.tokens.clone(),
-            tx_sender.ticker_requests.clone(),
+            tx_sender.ticker.clone(),
         ))
         .service(transaction::api_scope(tx_sender))
 }