text: use std::collections::{HashMap, VecDeque};
  // External uses
  use futures::{
      channel::{mpsc, oneshot},
      stream::StreamExt,
      SinkExt,
  };
  use tokio::task::JoinHandle;
  use web3::types::Address;
  // Workspace uses
  use crypto_exports::ff;
  use models::{
      node::{
          block::{
              Block, ExecutedOperations, ExecutedPriorityOp, ExecutedTx,
              PendingBlock as SendablePendingBlock,
          },
          mempool::SignedTxVariant,
          tx::{FranklinTx, TxHash},
          Account, AccountId, AccountTree, AccountUpdate, AccountUpdates, BlockNumber, PriorityOp,
      },
      ActionType, BlockCommitRequest, CommitRequest,
  };
  use plasma::state::{CollectedFee, OpSuccess, PlasmaState};
  use storage::ConnectionPool;
  // Local uses
  use crate::{gas_counter::GasCounter, mempool::ProposedBlock};
  use models::node::SignedFranklinTx;
  
  pub enum ExecutedOpId {
      Transaction(TxHash),
      PriorityOp(u64),
  }
  
  pub enum StateKeeperRequest {
      GetAccount(Address, oneshot::Sender<Option<(AccountId, Account)>>),
      GetLastUnprocessedPriorityOp(oneshot::Sender<u64>),
      ExecuteMiniBlock(ProposedBlock),
      GetExecutedInPendingBlock(
          ExecutedOpId,
          oneshot::Sender<Option<(BlockNumber, bool, Option<String>)>>,
      ),
      SealBlock,
  }
  
  pub struct ExecutedOpsNotify {
      pub operations: Vec<ExecutedOperations>,
      pub block_number: BlockNumber,
  }
  
  #[derive(Debug)]
  struct PendingBlock {
      success_operations: Vec<ExecutedOperations>,
      failed_txs: Vec<ExecutedTx>,
      account_updates: AccountUpdates,
      chunks_left: usize,
      pending_op_block_index: u32,
      unprocessed_priority_op_before: u64,
      pending_block_iteration: usize,
      withdrawals_amount: u32,
      gas_counter: GasCounter,
      /// Option denoting if this block should be generated faster than usual.
      fast_processing_required: bool,
      /// Fee should be applied only when sealing the block (because of corresponding logic in the circuit)
      collected_fees: Vec<CollectedFee>,
  }
  
  impl PendingBlock {
      fn new(unprocessed_priority_op_before: u64, chunks_left: usize) -> Self {
          Self {
              success_operations: Vec::new(),
              failed_txs: Vec::new(),
              account_updates: Vec::new(),
              chunks_left,
              pending_op_block_index: 0,
              unprocessed_priority_op_before,
              pending_block_iteration: 0,
              withdrawals_amount: 0,
              gas_counter: GasCounter::new(),
              fast_processing_required: false,
              collected_fees: Vec::new(),
          }
      }
  }
  
  /// Responsible for tx processing and block forming.
  pub struct PlasmaStateKeeper {
      /// Current plasma state
      state: PlasmaState,
  
      fee_account_id: AccountId,
      current_unprocessed_priority_op: u64,
  
      pending_block: PendingBlock,
  
      rx_for_blocks: mpsc::Receiver<StateKeeperRequest>,
      tx_for_commitments: mpsc::Sender<CommitRequest>,
      executed_tx_notify_sender: mpsc::Sender<ExecutedOpsNotify>,
  
      available_block_chunk_sizes: Vec<usize>,
      max_miniblock_iterations: usize,
      fast_miniblock_iterations: usize,
      max_number_of_withdrawals_per_block: usize,
  }
  
  pub struct PlasmaStateInitParams {
      pub tree: AccountTree,
      pub acc_id_by_addr: HashMap<Address, AccountId>,
      pub last_block_number: BlockNumber,
      pub unprocessed_priority_op: u64,
  }
  
  impl Default for PlasmaStateInitParams {
      fn default() -> Self {
          Self::new()
      }
  }
  
  impl PlasmaStateInitParams {
      pub fn new() -> Self {
          Self {
              tree: AccountTree::new(models::params::account_tree_depth()),
              acc_id_by_addr: HashMap::new(),
              last_block_number: 0,
              unprocessed_priority_op: 0,
          }
      }
  
      pub async fn get_pending_block(
          &self,
          storage: &mut storage::StorageProcessor<'_>,
      ) -> Option<SendablePendingBlock> {
          let pending_block = storage
              .chain()
              .block_schema()
              .load_pending_block()
              .await
              .unwrap_or_default()?;
  
          if pending_block.number <= self.last_block_number {
              // If after generating several pending block node generated
              // full blocks, they may be sealed on the first iteration
              // and stored pending block will be outdated.
              // Thus, if the stored pending block has the lower number than
              // last committed one, we just ignore it.
              return None;
          }
  
          // We've checked that pending block is greater than the last committed block,
          // but it must be greater exactly by 1.
          assert_eq!(pending_block.number, self.last_block_number + 1);
  
          Some(pending_block)
      }
  
      pub async fn restore_from_db(
          storage: &mut storage::StorageProcessor<'_>,
      ) -> Result<Self, failure::Error> {
          let mut init_params = Self::new();
          init_params.load_from_db(storage).await?;
  
          Ok(init_params)
      }
  
      async fn load_account_tree(
          &mut self,
          storage: &mut storage::StorageProcessor<'_>,
      ) -> Result<BlockNumber, failure::Error> {
          let (last_cached_block_number, accounts) = if let Some((block, _)) = storage
              .chain()
              .block_schema()
              .get_account_tree_cache()
              .await?
          {
              storage
                  .chain()
                  .state_schema()
                  .load_committed_state(Some(block))
                  .await?
          } else {
              storage.chain().state_schema().load_verified_state().await?
          };
  
          for (id, account) in accounts {
              self.insert_account(id, account);
          }
  
          if let Some(account_tree_cache) = storage
              .chain()
              .block_schema()
              .get_account_tree_cache_block(last_cached_block_number)
              .await?
          {
              self.tree
                  .set_internals(serde_json::from_value(account_tree_cache)?);
          } else {
              self.tree.root_hash();
              let account_tree_cache = self.tree.get_internals();
              storage
                  .chain()
                  .block_schema()
                  .store_account_tree_cache(
                      last_cached_block_number,
                      serde_json::to_value(account_tree_cache)?,
                  )
                  .await?;
          }
  
          let (block_number, accounts) = storage
              .chain()
              .state_schema()
              .load_committed_state(None)
              .await
              .map_err(|e| failure::format_err!("couldn't load committed state: {}", e))?;
  
          if block_number != last_cached_block_number {
              if let Some((_, account_updates)) = storage
                  .chain()
                  .state_schema()
                  .load_state_diff(last_cached_block_number, Some(block_number))
                  .await?
              {
                  let mut updated_accounts = account_updates
                      .into_iter()
                      .map(|(id, _)| id)
                      .collect::<Vec<_>>();
                  updated_accounts.sort();
                  updated_accounts.dedup();
                  for idx in updated_accounts {
                      if let Some(acc) = accounts.get(&idx).cloned() {
                          self.insert_account(idx, acc);
                      } else {
                          self.remove_account(idx);
                      }
                  }
              }
          }
          if block_number != 0 {
              let storage_root_hash = storage
                  .chain()
                  .block_schema()
                  .get_block(block_number)
                  .await?
                  .expect("restored block must exist");
              assert_eq!(
                  storage_root_hash.new_root_hash,
                  self.tree.root_hash(),
                  "restored root_hash is different"
              );
          }
          Ok(block_number)
      }
  
      async fn load_from_db(
          &mut self,
          storage: &mut storage::StorageProcessor<'_>,
      ) -> Result<(), failure::Error> {
          let block_number = self.load_account_tree(storage).await?;
          self.last_block_number = block_number;
          self.unprocessed_priority_op =
              Self::unprocessed_priority_op_id(storage, block_number).await?;
  
          info!(
              "Loaded committed state: last block number: {}, unprocessed priority op: {}",
              self.last_block_number, self.unprocessed_priority_op
          );
          Ok(())
      }
  
      pub async fn load_state_diff(
          &mut self,
          storage: &mut storage::StorageProcessor<'_>,
      ) -> Result<(), failure::Error> {
          let state_diff = storage
              .chain()
              .state_schema()
              .load_state_diff(self.last_block_number, None)
              .await
              .map_err(|e| failure::format_err!("failed to load committed state: {}", e))?;
  
          if let Some((block_number, updates)) = state_diff {
              for (id, update) in updates.into_iter() {
                  let updated_account = Account::apply_update(self.remove_account(id), update);
                  if let Some(account) = updated_account {
                      self.insert_account(id, account);
                  }
              }
              self.unprocessed_priority_op =
                  Self::unprocessed_priority_op_id(storage, block_number).await?;
              self.last_block_number = block_number;
          }
          Ok(())
      }
  
      pub fn insert_account(&mut self, id: u32, acc: Account) {
          self.acc_id_by_addr.insert(acc.address, id);
          self.tree.insert(id, acc);
      }
  
      pub fn remove_account(&mut self, id: u32) -> Option<Account> {
          if let Some(acc) = self.tree.remove(id) {
              self.acc_id_by_addr.remove(&acc.address);
              Some(acc)
          } else {
              None
          }
      }
  
      async fn unprocessed_priority_op_id(
          storage: &mut storage::StorageProcessor<'_>,
          block_number: BlockNumber,
      ) -> Result<u64, failure::Error> {
          let storage_op = storage
              .chain()
              .operations_schema()
              .get_operation(block_number, ActionType::COMMIT)
              .await;
          if let Some(storage_op) = storage_op {
              Ok(storage_op
                  .into_op(storage)
                  .await
                  .map_err(|e| failure::format_err!("could not convert storage_op: {}", e))?
                  .block
                  .processed_priority_ops
                  .1)
          } else {
              Ok(0)
          }
      }
  }
  
  impl PlasmaStateKeeper {
      #[allow(clippy::too_many_arguments)]
      pub fn new(
          initial_state: PlasmaStateInitParams,
          fee_account_address: Address,
          rx_for_blocks: mpsc::Receiver<StateKeeperRequest>,
          tx_for_commitments: mpsc::Sender<CommitRequest>,
          executed_tx_notify_sender: mpsc::Sender<ExecutedOpsNotify>,
          available_block_chunk_sizes: Vec<usize>,
          max_miniblock_iterations: usize,
          fast_miniblock_iterations: usize,
          max_number_of_withdrawals_per_block: usize,
      ) -> Self {
          assert!(!available_block_chunk_sizes.is_empty());
  
          let is_sorted = {
              let mut sorted = available_block_chunk_sizes.clone();
              sorted.sort();
              sorted == available_block_chunk_sizes
          };
          assert!(is_sorted);
  
          let state = PlasmaState::new(
              initial_state.tree,
              initial_state.acc_id_by_addr,
              initial_state.last_block_number + 1,
          );
  
          let (fee_account_id, _) = state
              .get_account_by_address(&fee_account_address)
              .expect("Fee account should be present in the account tree");
          // Keeper starts with the NEXT block
          let max_block_size = *available_block_chunk_sizes.iter().max().unwrap();
          let keeper = PlasmaStateKeeper {
              state,
              fee_account_id,
              current_unprocessed_priority_op: initial_state.unprocessed_priority_op,
              rx_for_blocks,
              tx_for_commitments,
              pending_block: PendingBlock::new(initial_state.unprocessed_priority_op, max_block_size),
              executed_tx_notify_sender,
              available_block_chunk_sizes,
              max_miniblock_iterations,
              fast_miniblock_iterations,
              max_number_of_withdrawals_per_block,
          };
  
          let root = keeper.state.root_hash();
          info!("created state keeper, root hash = {}", root);
  
          keeper
      }
  
      pub async fn initialize(&mut self, pending_block: Option<SendablePendingBlock>) {
          if let Some(pending_block) = pending_block {
              // Transform executed operations into non-executed, so they will be executed again.
              // Since it's a pending block, the state updates were not actually applied in the
              // database (as it happens only when full block is committed).
              //
              // We use `apply_tx` and `apply_priority_op` methods directly instead of
              // `apply_txs_batch` to preserve the original execution order. Otherwise there may
              // be a state corruption, if e.g. `Deposit` will be executed before `TransferToNew`
              // and account IDs will change.
              let mut txs_count = 0;
              let mut priority_op_count = 0;
              for operation in pending_block.success_operations {
                  match operation {
                      ExecutedOperations::Tx(tx) => {
                          self.apply_tx(&tx.signed_tx)
                              .expect("Tx from the restored pending block was not executed");
                          txs_count += 1;
                      }
                      ExecutedOperations::PriorityOp(op) => {
                          self.apply_priority_op(op.priority_op)
                              .expect("Priority op from the restored pending block was not executed");
                          priority_op_count += 1;
                      }
                  }
              }
  
              log::info!(
                  "Executed restored proposed block: {} transactions, {} priority operations, {} failed transactions",
                  txs_count,
                  priority_op_count,
                  pending_block.failed_txs.len()
              );
              self.pending_block.failed_txs = pending_block.failed_txs;
          } else {
              log::info!("There is no pending block to restore");
          }
      }
  
      pub async fn create_genesis_block(pool: ConnectionPool, fee_account_address: &Address) {
          let mut storage = pool
              .access_storage()
              .await
              .expect("db connection failed for statekeeper");
          let mut transaction = storage
              .start_transaction()
              .await
              .expect("unable to create db transaction in statekeeper");
  
          let (last_committed, mut accounts) = transaction
              .chain()
              .state_schema()
              .load_committed_state(None)
              .await
              .expect("db failed");
          // TODO: move genesis block creation to separate routine.
          assert!(
              last_committed == 0 && accounts.is_empty(),
              "db should be empty"
          );
          let mut fee_account = Account::default();
          fee_account.address = *fee_account_address;
          let db_account_update = AccountUpdate::Create {
              address: *fee_account_address,
              nonce: fee_account.nonce,
          };
          accounts.insert(0, fee_account);
          transaction
              .chain()
              .state_schema()
              .commit_state_update(0, &[(0, db_account_update)])
              .await
              .expect("db fail");
          transaction
              .chain()
              .state_schema()
              .apply_state_update(0)
              .await
              .expect("db fail");
  
          transaction
              .commit()
              .await
              .expect("Unable to commit transaction in statekeeper");
          let state = PlasmaState::from_acc_map(accounts, last_committed + 1);
          let root_hash = state.root_hash();
          info!("Genesis block created, state: {}", state.root_hash());
          println!("GENESIS_ROOT=0x{}", ff::to_hex(&root_hash));
      }
  
      async fn run(mut self, pending_block: Option<SendablePendingBlock>) {
          self.initialize(pending_block).await;
  
          let mut last_request_processed = std::time::Instant::now();
  
          while let Some(req) = self.rx_for_blocks.next().await {
              let start = std::time::Instant::now();
  
              log::trace!(
                  "Received new request. Last request was processed {}ms ago",
                  (start - last_request_processed).as_millis()
              );
  
              match req {
                  StateKeeperRequest::GetAccount(addr, sender) => {
                      sender.send(self.account(&addr)).unwrap_or_default();
  
                      log::trace!(
                          "GetAccount request processed in {}ms",
                          start.elapsed().as_millis()
                      );
                  }
                  StateKeeperRequest::GetLastUnprocessedPriorityOp(sender) => {
                      sender
                          .send(self.current_unprocessed_priority_op)
                          .unwrap_or_default();
  
                      log::trace!(
                          "GetLastUnprocessedPriorityOp request processed in {}ms",
                          start.elapsed().as_millis()
                      );
                  }
                  StateKeeperRequest::ExecuteMiniBlock(proposed_block) => {
                      self.execute_proposed_block(proposed_block).await;
  
                      log::trace!(
                          "ExecuteMiniBlock request processed in {}ms",
                          start.elapsed().as_millis()
                      );
                  }
                  StateKeeperRequest::GetExecutedInPendingBlock(op_id, sender) => {
                      let result = self.check_executed_in_pending_block(op_id);
                      sender.send(result).unwrap_or_default();
  
                      log::trace!(
                          "GetExecutedInPendingBlock request processed in {}ms",
                          start.elapsed().as_millis()
                      );
                  }
                  StateKeeperRequest::SealBlock => {
                      self.seal_pending_block().await;
  
                      log::trace!(
                          "SealBlock request processed in {}ms",
                          start.elapsed().as_millis()
                      );
                  }
              }
  
              last_request_processed = std::time::Instant::now();
          }
      }
  
      async fn notify_executed_ops(&self, executed_ops: &mut Vec<ExecutedOperations>) {
          if !executed_ops.is_empty() {
              self.executed_tx_notify_sender
                  .clone()
                  .send(ExecutedOpsNotify {
                      operations: executed_ops.clone(),
                      block_number: self.state.block_number,
                  })
                  .await
                  .map_err(|e| warn!("Failed to send executed tx notify batch: {}", e))
                  .unwrap_or_default();
          }
          executed_ops.clear();
      }
  
      async fn execute_proposed_block(&mut self, proposed_block: ProposedBlock) {
          let mut executed_ops = Vec::new();
  
          let mut priority_op_queue = proposed_block
              .priority_ops
              .into_iter()
              .collect::<VecDeque<_>>();
          while let Some(priority_op) = priority_op_queue.pop_front() {
              match self.apply_priority_op(priority_op) {
                  Ok(exec_op) => {
                      executed_ops.push(exec_op);
                  }
                  Err(priority_op) => {
                      self.seal_pending_block().await;
                      self.notify_executed_ops(&mut executed_ops).await;
  
                      priority_op_queue.push_front(priority_op);
                  }
              }
          }
  
          let mut tx_queue = proposed_block.txs.into_iter().collect::<VecDeque<_>>();
          while let Some(variant) = tx_queue.pop_front() {
              match &variant {
                  SignedTxVariant::Tx(tx) => {
                      match self.apply_tx(tx) {
                          Ok(exec_op) => {
                              executed_ops.push(exec_op);
                          }
                          Err(_) => {
                              // We could not execute the tx due to either of block size limit
                              // or the withdraw operations limit, so we seal this block and
                              // the last transaction will go to the next block instead.
                              self.seal_pending_block().await;
                              self.notify_executed_ops(&mut executed_ops).await;
  
                              tx_queue.push_front(variant);
                          }
                      }
                  }
                  SignedTxVariant::Batch(batch) => {
                      match self.apply_batch(&batch.txs, batch.batch_id) {
                          Ok(mut ops) => {
                              executed_ops.append(&mut ops);
                          }
                          Err(_) => {
                              // We could not execute the batch tx due to either of block size limit
                              // or the withdraw operations limit, so we seal this block and
                              // the last transaction will go to the next block instead.
                              self.seal_pending_block().await;
                              self.notify_executed_ops(&mut executed_ops).await;
  
                              tx_queue.push_front(variant);
                          }
                      }
                  }
              }
          }
  
          if !self.pending_block.success_operations.is_empty() {
              self.pending_block.pending_block_iteration += 1;
          }
  
          // If pending block contains withdrawals we seal it faster
          let max_miniblock_iterations = if self.pending_block.fast_processing_required {
              self.fast_miniblock_iterations
          } else {
              self.max_miniblock_iterations
          };
          if self.pending_block.pending_block_iteration > max_miniblock_iterations {
              self.seal_pending_block().await;
          } else {
              self.store_pending_block().await;
          }
  
          self.notify_executed_ops(&mut executed_ops).await;
      }
  
      // Err if there is no space in current block
      fn apply_priority_op(
          &mut self,
          priority_op: PriorityOp,
      ) -> Result<ExecutedOperations, PriorityOp> {
          let chunks_needed = priority_op.data.chunks();
          if self.pending_block.chunks_left < chunks_needed {
              return Err(priority_op);
          }
  
          // Check if adding this transaction to the block won't make the contract operations
          // too expensive.
          let non_executed_op = self
              .state
              .priority_op_to_franklin_op(priority_op.data.clone());
          if self
              .pending_block
              .gas_counter
              .add_op(&non_executed_op)
              .is_err()
          {
              // We've reached the gas limit, seal the block.
              // This transaction will go into the next one.
              return Err(priority_op);
          }
  
          let OpSuccess {
              fee,
              mut updates,
              executed_op,
          } = self.state.execute_priority_op(priority_op.data.clone());
  
          self.pending_block.chunks_left -= chunks_needed;
          self.pending_block.account_updates.append(&mut updates);
          if let Some(fee) = fee {
              self.pending_block.collected_fees.push(fee);
          }
          let block_index = self.pending_block.pending_op_block_index;
          self.pending_block.pending_op_block_index += 1;
  
          let exec_result = ExecutedOperations::PriorityOp(Box::new(ExecutedPriorityOp {
              op: executed_op,
              priority_op,
              block_index,
              created_at: chrono::Utc::now(),
          }));
          self.pending_block
              .success_operations
              .push(exec_result.clone());
          self.current_unprocessed_priority_op += 1;
          Ok(exec_result)
      }
  
      fn apply_batch(
          &mut self,
          txs: &[SignedFranklinTx],
          batch_id: i64,
      ) -> Result<Vec<ExecutedOperations>, ()> {
          let chunks_needed = self.state.chunks_for_batch(txs);
  
          // If we can't add the tx to the block due to the size limit, we return this tx,
          // seal the block and execute it again.
          if self.pending_block.chunks_left < chunks_needed {
              return Err(());
          }
  
          for tx in txs {
              // Check if adding this transaction to the block won't make the contract operations
              // too expensive.
              let non_executed_op = self.state.franklin_tx_to_franklin_op(tx.tx.clone());
              if let Ok(non_executed_op) = non_executed_op {
                  // We only care about successful conversions, since if conversion failed,
                  // then transaction will fail as well (as it shares the same code base).
                  if self
                      .pending_block
                      .gas_counter
                      .add_op(&non_executed_op)
                      .is_err()
                  {
                      // We've reached the gas limit, seal the block.
                      // This transaction will go into the next one.
                      return Err(());
                  }
              }
  
              if matches!(&tx.tx, &FranklinTx::Withdraw(_)) {
                  // Increase amount of the withdraw operations in this block.
                  self.pending_block.withdrawals_amount += 1;
              }
  
              // Check if we've reached the withdraw operations amount limit.
              // If so, this block will be sealed and this tx will go to the next block.
              if self.pending_block.withdrawals_amount
                  > self.max_number_of_withdrawals_per_block as u32
              {
                  return Err(());
              }
          }
  
          let all_updates = self.state.execute_txs_batch(txs);
          let mut executed_operations = Vec::new();
  
          for (tx, tx_updates) in txs.iter().zip(all_updates) {
              match tx_updates {
                  Ok(OpSuccess {
                      fee,
                      mut updates,
                      executed_op,
                  }) => {
                      self.pending_block.chunks_left -= chunks_needed;
                      self.pending_block.account_updates.append(&mut updates);
                      if let Some(fee) = fee {
                          let fee_updates = self.state.collect_fee(&[fee], self.fee_account_id);
                          self.pending_block
                              .account_updates
                              .extend(fee_updates.into_iter());
                      }
                      let block_index = self.pending_block.pending_op_block_index;
                      self.pending_block.pending_op_block_index += 1;
  
                      let exec_result = ExecutedOperations::Tx(Box::new(ExecutedTx {
                          signed_tx: tx.clone(),
                          success: true,
                          op: Some(executed_op),
                          fail_reason: None,
                          block_index: Some(block_index),
                          created_at: chrono::Utc::now(),
                          batch_id: Some(batch_id),
                      }));
                      self.pending_block
                          .success_operations
                          .push(exec_result.clone());
                      executed_operations.push(exec_result);
                  }
                  Err(e) => {
                      warn!("Failed to execute transaction: {:?}, {}", tx, e);
                      let failed_tx = ExecutedTx {
                          signed_tx: tx.clone(),
                          success: false,
                          op: None,
                          fail_reason: Some(e.to_string()),
                          block_index: None,
                          created_at: chrono::Utc::now(),
                          batch_id: Some(batch_id),
                      };
                      self.pending_block.failed_txs.push(failed_tx.clone());
                      let exec_result = ExecutedOperations::Tx(Box::new(failed_tx));
                      executed_operations.push(exec_result);
                  }
              };
          }
  
          Ok(executed_operations)
      }
  
      fn apply_tx(&mut self, tx: &SignedFranklinTx) -> Result<ExecutedOperations, ()> {
          let chunks_needed = self.state.chunks_for_tx(&tx);
  
          // If we can't add the tx to the block due to the size limit, we return this tx,
          // seal the block and execute it again.
          if self.pending_block.chunks_left < chunks_needed {
              return Err(());
          }
  
          // Check if adding this transaction to the block won't make the contract operations
          // too expensive.
          let non_executed_op = self.state.franklin_tx_to_franklin_op(tx.tx.clone());
          if let Ok(non_executed_op) = non_executed_op {
              // We only care about successful conversions, since if conversion failed,
              // then transaction will fail as well (as it shares the same code base).
              if self
                  .pending_block
                  .gas_counter
                  .add_op(&non_executed_op)
                  .is_err()
              {
                  // We've reached the gas limit, seal the block.
                  // This transaction will go into the next one.
                  return Err(());
              }
          }
  
          if let FranklinTx::Withdraw(tx) = &tx.tx {
              // Increase amount of the withdraw operations in this block.
              self.pending_block.withdrawals_amount += 1;
  
              // Check if we should mark this block as requiring fast processing.
              if tx.fast {
                  self.pending_block.fast_processing_required = true;
              }
          }
  
          // Check if we've reached the withdraw operations amount limit.
          // If so, this block will be sealed and this tx will go to the next block.
          if self.pending_block.withdrawals_amount > self.max_number_of_withdrawals_per_block as u32 {
              return Err(());
          }
  
          let tx_updates = self.state.execute_tx(tx.tx.clone());
  
          let exec_result = match tx_updates {
              Ok(OpSuccess {
                  fee,
                  mut updates,
                  executed_op,
              }) => {
                  self.pending_block.chunks_left -= chunks_needed;
                  self.pending_block.account_updates.append(&mut updates);
                  if let Some(fee) = fee {
                      self.pending_block.collected_fees.push(fee);
                  }
                  let block_index = self.pending_block.pending_op_block_index;
                  self.pending_block.pending_op_block_index += 1;
  
                  let exec_result = ExecutedOperations::Tx(Box::new(ExecutedTx {
                      signed_tx: tx.clone(),
                      success: true,
                      op: Some(executed_op),
                      fail_reason: None,
                      block_index: Some(block_index),
                      created_at: chrono::Utc::now(),
                      batch_id: None,
                  }));
                  self.pending_block
                      .success_operations
                      .push(exec_result.clone());
                  exec_result
              }
              Err(e) => {
                  warn!("Failed to execute transaction: {:?}, {}", tx, e);
                  let failed_tx = ExecutedTx {
                      signed_tx: tx.clone(),
                      success: false,
                      op: None,
                      fail_reason: Some(e.to_string()),
                      block_index: None,
                      created_at: chrono::Utc::now(),
                      batch_id: None,
                  };
                  self.pending_block.failed_txs.push(failed_tx.clone());
                  ExecutedOperations::Tx(Box::new(failed_tx))
              }
          };
  
          Ok(exec_result)
      }
  
      /// Finalizes the pending block, transforming it into a full block.
      async fn seal_pending_block(&mut self) {
          let mut pending_block = std::mem::replace(
              &mut self.pending_block,
              PendingBlock::new(
                  self.current_unprocessed_priority_op,
                  *self
                      .available_block_chunk_sizes
                      .last()
                      .expect("failed to get max block size"),
              ),
          );
  
          // Apply fees of pending block
          let fee_updates = self
              .state
              .collect_fee(&pending_block.collected_fees, self.fee_account_id);
          pending_block
              .account_updates
              .extend(fee_updates.into_iter());
  
          let mut block_transactions = pending_block.success_operations;
          block_transactions.extend(
              pending_block
                  .failed_txs
                  .into_iter()
                  .map(|tx| ExecutedOperations::Tx(Box::new(tx))),
          );
  
          let commit_gas_limit = pending_block.gas_counter.commit_gas_limit();
          let verify_gas_limit = pending_block.gas_counter.verify_gas_limit();
  
          let block_commit_request = BlockCommitRequest {
              block: Block::new_from_availabe_block_sizes(
                  self.state.block_number,
                  self.state.root_hash(),
                  self.fee_account_id,
                  block_transactions,
                  (
                      pending_block.unprocessed_priority_op_before,
                      self.current_unprocessed_priority_op,
                  ),
                  &self.available_block_chunk_sizes,
                  commit_gas_limit,
                  verify_gas_limit,
              ),
              accounts_updated: pending_block.account_updates,
          };
          self.state.block_number += 1;
  
          info!(
              "Creating full block: {}, operations: {}, chunks_left: {}, miniblock iterations: {}",
              block_commit_request.block.block_number,
              block_commit_request.block.block_transactions.len(),
              pending_block.chunks_left,
              pending_block.pending_block_iteration
          );
  
          let (notification_sender, notification_receiver) = oneshot::channel::<()>();
  
          let commit_request = CommitRequest::Block(block_commit_request, notification_sender);
          self.tx_for_commitments
              .send(commit_request)
              .await
              .expect("committer receiver dropped");
  
          notification_receiver
              .await
              .expect("committer sender dropped");
      }
  
      /// Stores intermediate representation of a pending block in the database,
      /// so the executed transactions are persisted and won't be lost.
      async fn store_pending_block(&mut self) {
          // Create a pending block object to send.
          // Note that failed operations are not included, as per any operation failure
          // the full block is created immediately.
          let pending_block = SendablePendingBlock {
              number: self.state.block_number,
              chunks_left: self.pending_block.chunks_left,
              unprocessed_priority_op_before: self.pending_block.unprocessed_priority_op_before,
              pending_block_iteration: self.pending_block.pending_block_iteration,
              success_operations: self.pending_block.success_operations.clone(),
              failed_txs: self.pending_block.failed_txs.clone(),
          };
  
          log::trace!(
              "Persisting mini block: {}, operations: {}, failed_txs: {}, chunks_left: {}, miniblock iterations: {}",
              pending_block.number,
              pending_block.success_operations.len(),
              pending_block.failed_txs.len(),
              pending_block.chunks_left,
              pending_block.pending_block_iteration
          );
  
          let (notification_sender, notification_receiver) = oneshot::channel::<()>();
  
          let commit_request = CommitRequest::PendingBlock(pending_block, notification_sender);
          self.tx_for_commitments
              .send(commit_request)
              .await
              .expect("committer receiver dropped");
  
          notification_receiver
              .await
              .expect("committer sender dropped");
      }
  
      fn check_executed_in_pending_block(
          &self,
          op_id: ExecutedOpId,
      ) -> Option<(BlockNumber, bool, Option<String>)> {
          let current_block_number = self.state.block_number;
          match op_id {
              ExecutedOpId::Transaction(hash) => {
                  for op in &self.pending_block.success_operations {
                      if let ExecutedOperations::Tx(exec_tx) = op {
                          if exec_tx.signed_tx.hash() == hash {
                              return Some((current_block_number, true, None));
                          }
                      }
                  }
  
                  for failed_tx in &self.pending_block.failed_txs {
                      if failed_tx.signed_tx.hash() == hash {
                          return Some((current_block_number, false, failed_tx.fail_reason.clone()));
                      }
                  }
              }
              ExecutedOpId::PriorityOp(serial_id) => {
                  for op in &self.pending_block.success_operations {
                      if let ExecutedOperations::PriorityOp(exec_op) = op {
                          if exec_op.priority_op.serial_id == serial_id {
                              return Some((current_block_number, true, None));
                          }
                      }
                  }
              }
          }
          None
      }
  
      fn account(&self, address: &Address) -> Option<(AccountId, Account)> {
          self.state.get_account_by_address(address)
      }
  }
  
  #[must_use]
  pub fn start_state_keeper(
      sk: PlasmaStateKeeper,
      pending_block: Option<SendablePendingBlock>,
  ) -> JoinHandle<()> {
      tokio::spawn(sk.run(pending_block))
  }
  