Restore fast withdrawals

core/bin/zksync_core/src/committer/aggregated_committer.rs

@@ -20,6 +20,7 @@ fn create_new_commit_operation(
     max_blocks_to_commit: usize,
     block_commit_deadline: Duration,
     max_gas_for_tx: U256,
+    fast_processing: bool,
 ) -> Option<BlocksCommitOperation> {
     let new_blocks = new_blocks
         .iter()
@@ -45,7 +46,8 @@ fn create_new_commit_operation(
 
     let should_commit_blocks = any_block_commit_deadline_triggered
         || gas_limit_reached_for_blocks
-        || new_blocks.len() == max_blocks_to_commit;
+        || new_blocks.len() == max_blocks_to_commit
+        || fast_processing;
     if !should_commit_blocks {
         return None;
     }
@@ -73,6 +75,7 @@ fn create_new_create_proof_operation(
     current_time: DateTime<Utc>,
     block_verify_deadline: Duration,
     _max_gas_for_tx: U256,
+    fast_processing: bool,
 ) -> Option<BlocksCreateProofOperation> {
     let max_aggregate_size = available_aggregate_proof_sizes
         .last()
@@ -98,7 +101,7 @@ fn create_new_create_proof_operation(
     let can_create_max_aggregate_proof = new_blocks_with_proofs.len() >= max_aggregate_size;
 
     let should_create_aggregate_proof =
-        any_block_verify_deadline_triggered || can_create_max_aggregate_proof;
+        any_block_verify_deadline_triggered || can_create_max_aggregate_proof || fast_processing;
 
     if !should_create_aggregate_proof {
         return None;
@@ -146,6 +149,7 @@ fn create_execute_blocks_operation(
     max_blocks_to_execute: usize,
     block_execute_deadline: Duration,
     max_gas_for_tx: U256,
+    fast_processing: bool,
 ) -> Option<BlocksExecuteOperation> {
     let proven_non_executed_block = proven_non_executed_block
         .iter()
@@ -170,7 +174,8 @@ fn create_execute_blocks_operation(
 
     let should_execute_blocks = any_block_execute_deadline_triggered
         || gas_limit_reached_for_blocks
-        || proven_non_executed_block.len() == max_blocks_to_execute;
+        || proven_non_executed_block.len() == max_blocks_to_execute
+        || fast_processing;
     if !should_execute_blocks {
         return None;
     }
@@ -191,6 +196,27 @@ fn create_execute_blocks_operation(
     })
 }
 
+/// Checks if fast processing is required for any `Block`
+async fn is_fast_processing_requested(
+    storage: &mut StorageProcessor<'_>,
+    blocks: &[Block],
+) -> anyhow::Result<bool> {
+    let mut fast_processing = false;
+    for block in blocks {
+        let fast_processing_for_current_block_requested = BlockSchema(storage)
+            .get_block_metadata(block.block_number)
+            .await?
+            .map(|mdat| mdat.fast_processing)
+            .unwrap_or(false);
+
+        fast_processing = fast_processing || fast_processing_for_current_block_requested;
+        if fast_processing {
+            break;
+        }
+    }
+    return Ok(fast_processing);
+}
+
 async fn create_aggregated_commits_storage(
     storage: &mut StorageProcessor<'_>,
     config: &ZkSyncConfig,
@@ -211,13 +237,16 @@ async fn create_aggregated_commits_storage(
         block_number.0 += 1;
     }
 
+    let fast_processing_requested = is_fast_processing_requested(storage, &new_blocks).await?;
+
     let commit_operation = create_new_commit_operation(
         &old_committed_block,
         &new_blocks,
         Utc::now(),
         config.chain.state_keeper.max_aggregated_blocks_to_commit,
         config.chain.state_keeper.block_commit_deadline(),
         config.chain.state_keeper.max_aggregated_tx_gas.into(),
+        fast_processing_requested,
     );
 
     if let Some(commit_operation) = commit_operation {
@@ -264,12 +293,16 @@ async fn create_aggregated_prover_task_storage(
         }
     }
 
+    let fast_processing_requested =
+        is_fast_processing_requested(storage, &blocks_with_proofs).await?;
+
     let create_proof_operation = create_new_create_proof_operation(
         &blocks_with_proofs,
         &config.chain.state_keeper.aggregated_proof_sizes,
         Utc::now(),
         config.chain.state_keeper.block_prove_deadline(),
         config.chain.state_keeper.max_aggregated_tx_gas.into(),
+        fast_processing_requested,
     );
     if let Some(operation) = create_proof_operation {
         let aggregated_op = operation.into();
@@ -374,12 +407,15 @@ async fn create_aggregated_execute_operation_storage(
         blocks.push(block);
     }
 
+    let fast_processing_requested = is_fast_processing_requested(storage, &blocks).await?;
+
     let execute_operation = create_execute_blocks_operation(
         &blocks,
         Utc::now(),
         config.chain.state_keeper.max_aggregated_blocks_to_execute,
         config.chain.state_keeper.block_execute_deadline(),
         config.chain.state_keeper.max_aggregated_tx_gas.into(),
+        fast_processing_requested,
     );
 
     if let Some(operation) = execute_operation {

core/bin/zksync_core/src/committer/mod.rs

@@ -10,7 +10,7 @@ use crate::mempool::MempoolBlocksRequest;
 use zksync_config::ZkSyncConfig;
 use zksync_storage::ConnectionPool;
 use zksync_types::{
-    block::{Block, ExecutedOperations, PendingBlock},
+    block::{Block, BlockMetadata, ExecutedOperations, PendingBlock},
     AccountUpdates, BlockNumber,
 };
 
@@ -25,6 +25,7 @@ pub enum CommitRequest {
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct BlockCommitRequest {
     pub block: Block,
+    pub block_metadata: BlockMetadata,
     pub accounts_updated: AccountUpdates,
 }
 
@@ -127,6 +128,7 @@ async fn commit_block(
     let start = Instant::now();
     let BlockCommitRequest {
         block,
+        block_metadata,
         accounts_updated,
     } = block_commit_request;
 
@@ -165,13 +167,22 @@ async fn commit_block(
 
     vlog::info!("commit block #{}", block.block_number);
 
+    let block_number = block.block_number;
+
     transaction
         .chain()
         .block_schema()
         .save_block(block)
         .await
         .expect("committer must commit the op into db");
 
+    transaction
+        .chain()
+        .block_schema()
+        .save_block_metadata(block_number, block_metadata)
+        .await
+        .expect("committer must commit block block metadata into db");
+
     mempool_req_sender
         .send(MempoolBlocksRequest::UpdateNonces(accounts_updated))
         .await

core/bin/zksync_core/src/state_keeper/mod.rs

@@ -20,7 +20,7 @@ use zksync_state::state::{CollectedFee, OpSuccess, ZkSyncState};
 use zksync_storage::ConnectionPool;
 use zksync_types::{
     block::{
-        Block, ExecutedOperations, ExecutedPriorityOp, ExecutedTx,
+        Block, BlockMetadata, ExecutedOperations, ExecutedPriorityOp, ExecutedTx,
         PendingBlock as SendablePendingBlock,
     },
     gas_counter::GasCounter,
@@ -1004,8 +1004,13 @@ impl ZkSyncStateKeeper {
 
         self.pending_block.previous_block_root_hash = block.get_eth_encoded_root();
 
+        let block_metadata = BlockMetadata {
+            fast_processing: pending_block.fast_processing_required,
+        };
+
         let block_commit_request = BlockCommitRequest {
             block,
+            block_metadata,
             accounts_updated: pending_block.account_updates.clone(),
         };
         let first_update_order_id = pending_block.stored_account_updates;

core/lib/storage/migrations/2021-02-23-171907_fast_withdraw_aggregated_blocks/down.sql

@@ -0,0 +1 @@
+drop table block_metadata;

core/lib/storage/migrations/2021-02-23-171907_fast_withdraw_aggregated_blocks/up.sql

@@ -0,0 +1,5 @@
+create table block_metadata (
+    block_number bigserial NOT NULL REFERENCES blocks (number) on delete cascade,
+    fast_processing  boolean not null,
+    primary key (block_number)
+);

core/lib/storage/sqlx-data.json

@@ -1305,6 +1305,32 @@
       ]
     }
   },
+  "47dd80567908f3b37161e4f92a97654e7af4a5e921145bdedbc446a653926b88": {
+    "query": "SELECT * FROM block_metadata WHERE block_number = $1",
+    "describe": {
+      "columns": [
+        {
+          "ordinal": 0,
+          "name": "block_number",
+          "type_info": "Int8"
+        },
+        {
+          "ordinal": 1,
+          "name": "fast_processing",
+          "type_info": "Bool"
+        }
+      ],
+      "parameters": {
+        "Left": [
+          "Int8"
+        ]
+      },
+      "nullable": [
+        false,
+        false
+      ]
+    }
+  },
   "47e6a9e74f9281ef8f9373829fd8500920226c4a9ef3546b2d01fb0dfb20d686": {
     "query": "\n                SELECT aggregate_operations.* FROM eth_aggregated_ops_binding\n                LEFT JOIN aggregate_operations ON aggregate_operations.id = op_id\n                WHERE eth_op_id = $1\n                ",
     "describe": {
@@ -1980,6 +2006,19 @@
       ]
     }
   },
+  "73eedd4444ef5bfbfd526c319f97d75609a65517d63e88add0a864a9f7141a02": {
+    "query": "\n            INSERT INTO block_metadata (block_number, fast_processing)\n            VALUES ($1, $2)\n            ",
+    "describe": {
+      "columns": [],
+      "parameters": {
+        "Left": [
+          "Int8",
+          "Bool"
+        ]
+      },
+      "nullable": []
+    }
+  },
   "74a5cc4affa23433b5b7834df6dfa1a7a2c5a65f23289de3de5a4f1b93f89c06": {
     "query": "SELECT address FROM account_creates WHERE account_id = $1",
     "describe": {

core/lib/storage/src/chain/block/mod.rs

@@ -6,12 +6,13 @@ use zksync_basic_types::{H256, U256};
 use zksync_crypto::convert::FeConvert;
 use zksync_types::{
     aggregated_operations::AggregatedActionType,
-    block::{Block, ExecutedOperations, PendingBlock},
+    block::{Block, BlockMetadata, ExecutedOperations, PendingBlock},
     AccountId, BlockNumber, Fr, ZkSyncOp,
 };
 // Local imports
 use self::records::{
-    AccountTreeCache, BlockDetails, BlockTransactionItem, StorageBlock, StoragePendingBlock,
+    AccountTreeCache, BlockDetails, BlockTransactionItem, StorageBlock, StorageBlockMetadata,
+    StoragePendingBlock,
 };
 use crate::{
     chain::operations::{
@@ -122,6 +123,31 @@ impl<'a, 'c> BlockSchema<'a, 'c> {
         Ok(result)
     }
 
+    /// Given the block number, attempts to get metadata related to block.
+    /// Returns `None` if not found.
+    pub async fn get_block_metadata(
+        &mut self,
+        block: BlockNumber,
+    ) -> QueryResult<Option<BlockMetadata>> {
+        let start = Instant::now();
+
+        let db_result = sqlx::query_as!(
+            StorageBlockMetadata,
+            "SELECT * FROM block_metadata WHERE block_number = $1",
+            i64::from(*block)
+        )
+        .fetch_optional(self.0.conn())
+        .await?;
+
+        metrics::histogram!("sql.chain.block.get_block_metadata", start.elapsed());
+
+        let result = db_result.map(|md| BlockMetadata {
+            fast_processing: md.fast_processing,
+        });
+
+        Ok(result)
+    }
+
     /// Same as `get_block_executed_ops`, but returns a vector of `ZkSyncOp` instead
     /// of `ExecutedOperations`.
     pub async fn get_block_operations(&mut self, block: BlockNumber) -> QueryResult<Vec<ZkSyncOp>> {
@@ -681,6 +707,28 @@ impl<'a, 'c> BlockSchema<'a, 'c> {
         Ok(())
     }
 
+    pub async fn save_block_metadata(
+        &mut self,
+        block_number: BlockNumber,
+        block_metadata: BlockMetadata,
+    ) -> QueryResult<()> {
+        let start = Instant::now();
+
+        sqlx::query!(
+            "
+            INSERT INTO block_metadata (block_number, fast_processing)
+            VALUES ($1, $2)
+            ",
+            i64::from(*block_number),
+            block_metadata.fast_processing
+        )
+        .execute(self.0.conn())
+        .await?;
+
+        metrics::histogram!("sql.chain.block.save_block_metadata", start.elapsed());
+        Ok(())
+    }
+
     /// Stores account tree cache for a block
     pub async fn store_account_tree_cache(
         &mut self,

core/lib/storage/src/chain/block/records.rs

@@ -77,3 +77,9 @@ impl BlockDetails {
         self.verified_at.is_some() && self.verify_tx_hash.is_some()
     }
 }
+
+#[derive(Debug, FromRow)]
+pub struct StorageBlockMetadata {
+    pub block_number: i64,
+    pub fast_processing: bool,
+}

core/lib/types/src/block.rs

@@ -453,3 +453,9 @@ pub struct OnchainOperationsBlockInfo {
     pub public_data_offset: u32,
     pub eth_witness: Vec<u8>,
 }
+
+/// Additional data attached to block that is not related to the core protocol
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct BlockMetadata {
+    pub fast_processing: bool,
+}

infrastructure/zcli/src/commands.ts

@@ -203,7 +203,7 @@ class TxSubmitter {
     }
 
     private async withdraw(txDetails: TxDetails, fast: boolean) {
-        const { to: ethAddress, token, amount } = txDetails;
+        const { to: ethAddress, token, amount, fastProcessing } = txDetails;
         if (!(await this.syncWallet.isSigningKeySet())) {
             const changePubkey = await this.syncWallet.setSigningKey({
                 feeToken: token,
@@ -214,7 +214,8 @@ class TxSubmitter {
         const txHandle = await this.syncWallet.withdrawFromSyncToEthereum({
             ethAddress,
             token,
-            amount: this.syncProvider.tokenSet.parseToken(token, amount)
+            amount: this.syncProvider.tokenSet.parseToken(token, amount),
+            fastProcessing
         });
         if (!fast) await txHandle.awaitReceipt();
         return txHandle.txHash;