[backport] Transactional updates in the state-db (#1616) (#1626)

* Transactional updates in the state-db (#1616)

* set link-arg --export-table for wasm builds (#1598)

Author: arkpar

Cargo.lock

@@ -608,7 +608,8 @@ impl<Block: BlockT> Backend<Block> {
 			};
 
 			trace!(target: "db", "Canonicalize block #{} ({:?})", new_canonical, hash);
-			let commit = self.storage.state_db.canonicalize_block(&hash);
+			let commit = self.storage.state_db.canonicalize_block(&hash)
+				.map_err(|e: state_db::Error<io::Error>| client::error::Error::from(format!("State database error: {:?}", e)))?;
 			apply_state_commit(transaction, commit);
 		};
 
@@ -641,7 +642,8 @@ impl<Block: BlockT> Backend<Block> {
 			let lookup_key = ::utils::number_and_hash_to_lookup_key(f_num, f_hash.clone());
 			transaction.put(columns::META, meta_keys::FINALIZED_BLOCK, &lookup_key);
 
-			let commit = self.storage.state_db.canonicalize_block(&f_hash);
+			let commit = self.storage.state_db.canonicalize_block(&f_hash)
+				.map_err(|e: state_db::Error<io::Error>| client::error::Error::from(format!("State database error: {:?}", e)))?;
 			apply_state_commit(transaction, commit);
 
 			// read config from genesis, since it is readonly atm
@@ -654,67 +656,11 @@ impl<Block: BlockT> Backend<Block> {
 
 		Ok(())
 	}
-}
-
-fn apply_state_commit(transaction: &mut DBTransaction, commit: state_db::CommitSet<H256>) {
-	for (key, val) in commit.data.inserted.into_iter() {
-		transaction.put(columns::STATE, &key[..], &val);
-	}
-	for key in commit.data.deleted.into_iter() {
-		transaction.delete(columns::STATE, &key[..]);
-	}
-	for (key, val) in commit.meta.inserted.into_iter() {
-		transaction.put(columns::STATE_META, &key[..], &val);
-	}
-	for key in commit.meta.deleted.into_iter() {
-		transaction.delete(columns::STATE_META, &key[..]);
-	}
-}
-
-impl<Block> client::backend::AuxStore for Backend<Block> where Block: BlockT<Hash=H256> {
-	fn insert_aux<
-		'a,
-		'b: 'a,
-		'c: 'a,
-		I: IntoIterator<Item=&'a(&'c [u8], &'c [u8])>,
-		D: IntoIterator<Item=&'a &'b [u8]>,
-	>(&self, insert: I, delete: D) -> client::error::Result<()> {
-		let mut transaction = DBTransaction::new();
-		for (k, v) in insert {
-			transaction.put(columns::AUX, k, v);
-		}
-		for k in delete {
-			transaction.delete(columns::AUX, k);
-		}
-		self.storage.db.write(transaction).map_err(db_err)?;
-		Ok(())
-	}
-
-	fn get_aux(&self, key: &[u8]) -> Result<Option<Vec<u8>>, client::error::Error> {
-		Ok(self.storage.db.get(columns::AUX, key).map(|r| r.map(|v| v.to_vec())).map_err(db_err)?)
-	}
-}
-
-impl<Block> client::backend::Backend<Block, Blake2Hasher> for Backend<Block> where Block: BlockT<Hash=H256> {
-	type BlockImportOperation = BlockImportOperation<Block, Blake2Hasher>;
-	type Blockchain = BlockchainDb<Block>;
-	type State = CachingState<Blake2Hasher, DbState, Block>;
-	type ChangesTrieStorage = DbChangesTrieStorage<Block>;
 
-	fn begin_operation(&self, block: BlockId<Block>) -> Result<Self::BlockImportOperation, client::error::Error> {
-		let state = self.state_at(block)?;
-		Ok(BlockImportOperation {
-			pending_block: None,
-			old_state: state,
-			db_updates: MemoryDB::default(),
-			storage_updates: Default::default(),
-			changes_trie_updates: MemoryDB::default(),
-			aux_ops: Vec::new(),
-		})
-	}
-
-	fn commit_operation(&self, mut operation: Self::BlockImportOperation)
+	fn try_commit_operation(&self, mut operation: BlockImportOperation<Block, Blake2Hasher>)
 		-> Result<(), client::error::Error>
+		where
+			Block: BlockT<Hash=H256>,
 	{
 		let mut transaction = DBTransaction::new();
 		operation.apply_aux(&mut transaction);
@@ -747,7 +693,7 @@ impl<Block> client::backend::Backend<Block, Blake2Hasher> for Backend<Block> whe
 						retracted.push(r.hash.clone());
 						if r.hash == meta.finalized_hash {
 							warn!("Potential safety failure: reverting finalized block {:?}",
-								(&r.number, &r.hash));
+								  (&r.number, &r.hash));
 
 							return Err(::client::error::ErrorKind::NotInFinalizedChain.into());
 						}
@@ -812,22 +758,27 @@ impl<Block> client::backend::Backend<Block, Blake2Hasher> for Backend<Block> whe
 			let commit = self.storage.state_db.insert_block(&hash, number_u64, &pending_block.header.parent_hash(), changeset)
 				.map_err(|e: state_db::Error<io::Error>| client::error::Error::from(format!("State database error: {:?}", e)))?;
 			apply_state_commit(&mut transaction, commit);
-			self.changes_tries_storage.commit(&mut transaction, operation.changes_trie_updates);
 
 			let finalized = match pending_block.leaf_state {
 				NewBlockState::Final => true,
 				_ => false,
 			};
 
+			let header = &pending_block.header;
+			let is_best = pending_block.leaf_state.is_best();
+			let changes_trie_updates = operation.changes_trie_updates;
+
+
+			self.changes_tries_storage.commit(&mut transaction, changes_trie_updates);
+
 			if finalized {
 				// TODO: ensure best chain contains this block.
 				self.note_finalized(&mut transaction, &pending_block.header, hash)?;
 			} else {
 				// canonicalize blocks which are old enough, regardless of finality.
-				self.force_delayed_canonicalize(&mut transaction, hash, *pending_block.header.number())?
+				self.force_delayed_canonicalize(&mut transaction, hash, *header.number())?
 			}
 
-			let is_best = pending_block.leaf_state.is_best();
 			debug!(target: "db", "DB Commit {:?} ({}), best = {}", hash, number, is_best);
 
 			{
@@ -849,7 +800,7 @@ impl<Block> client::backend::Backend<Block, Blake2Hasher> for Backend<Block> whe
 			self.blockchain.update_meta(
 				hash.clone(),
 				number.clone(),
-				pending_block.leaf_state.is_best(),
+				is_best,
 				finalized,
 			);
 
@@ -865,31 +816,114 @@ impl<Block> client::backend::Backend<Block, Blake2Hasher> for Backend<Block> whe
 		}
 		Ok(())
 	}
+}
+
+fn apply_state_commit(transaction: &mut DBTransaction, commit: state_db::CommitSet<H256>) {
+	for (key, val) in commit.data.inserted.into_iter() {
+		transaction.put(columns::STATE, &key[..], &val);
+	}
+	for key in commit.data.deleted.into_iter() {
+		transaction.delete(columns::STATE, &key[..]);
+	}
+	for (key, val) in commit.meta.inserted.into_iter() {
+		transaction.put(columns::STATE_META, &key[..], &val);
+	}
+	for key in commit.meta.deleted.into_iter() {
+		transaction.delete(columns::STATE_META, &key[..]);
+	}
+}
+
+impl<Block> client::backend::AuxStore for Backend<Block> where Block: BlockT<Hash=H256> {
+	fn insert_aux<
+		'a,
+		'b: 'a,
+		'c: 'a,
+		I: IntoIterator<Item=&'a(&'c [u8], &'c [u8])>,
+		D: IntoIterator<Item=&'a &'b [u8]>,
+	>(&self, insert: I, delete: D) -> client::error::Result<()> {
+		let mut transaction = DBTransaction::new();
+		for (k, v) in insert {
+			transaction.put(columns::AUX, k, v);
+		}
+		for k in delete {
+			transaction.delete(columns::AUX, k);
+		}
+		self.storage.db.write(transaction).map_err(db_err)?;
+		Ok(())
+	}
+
+	fn get_aux(&self, key: &[u8]) -> Result<Option<Vec<u8>>, client::error::Error> {
+		Ok(self.storage.db.get(columns::AUX, key).map(|r| r.map(|v| v.to_vec())).map_err(db_err)?)
+	}
+}
+
+impl<Block> client::backend::Backend<Block, Blake2Hasher> for Backend<Block> where Block: BlockT<Hash=H256> {
+	type BlockImportOperation = BlockImportOperation<Block, Blake2Hasher>;
+	type Blockchain = BlockchainDb<Block>;
+	type State = CachingState<Blake2Hasher, DbState, Block>;
+	type ChangesTrieStorage = DbChangesTrieStorage<Block>;
+
+	fn begin_operation(&self, block: BlockId<Block>) -> Result<Self::BlockImportOperation, client::error::Error> {
+		let state = self.state_at(block)?;
+		Ok(BlockImportOperation {
+			pending_block: None,
+			old_state: state,
+			db_updates: MemoryDB::default(),
+			storage_updates: Default::default(),
+			changes_trie_updates: MemoryDB::default(),
+			aux_ops: Vec::new(),
+		})
+	}
+
+	fn commit_operation(&self, operation: Self::BlockImportOperation)
+		-> Result<(), client::error::Error>
+	{
+		match self.try_commit_operation(operation) {
+			Ok(_) => {
+				self.storage.state_db.apply_pending();
+				Ok(())
+			},
+			e @ Err(_) => {
+				self.storage.state_db.revert_pending();
+				e
+			}
+		}
+	}
 
 	fn finalize_block(&self, block: BlockId<Block>, justification: Option<Justification>)
 		-> Result<(), client::error::Error>
 	{
 		use runtime_primitives::traits::Header;
 
-		if let Some(header) = ::client::blockchain::HeaderBackend::header(&self.blockchain, block)? {
-			let mut transaction = DBTransaction::new();
-			// TODO: ensure best chain contains this block.
-			let hash = header.hash();
-			self.note_finalized(&mut transaction, &header, hash.clone())?;
-			if let Some(justification) = justification {
-				let number = header.number().clone();
-				transaction.put(
-					columns::JUSTIFICATION,
-					&::utils::number_and_hash_to_lookup_key(number, hash.clone()),
-					&justification.encode(),
-				);
+		let commit = || {
+			if let Some(header) = ::client::blockchain::HeaderBackend::header(&self.blockchain, block)? {
+				let mut transaction = DBTransaction::new();
+				// TODO: ensure best chain contains this block.
+				let hash = header.hash();
+				self.note_finalized(&mut transaction, &header, hash.clone())?;
+				if let Some(justification) = justification {
+					let number = header.number().clone();
+					transaction.put(
+						columns::JUSTIFICATION,
+						&::utils::number_and_hash_to_lookup_key(number, hash.clone()),
+						&justification.encode(),
+					);
+				}
+				self.storage.db.write(transaction).map_err(db_err)?;
+				self.blockchain.update_meta(hash, header.number().clone(), false, true);
+				Ok(())
+			} else {
+				Err(client::error::ErrorKind::UnknownBlock(format!("Cannot finalize block {:?}", block)).into())
+			}
+		};
+		match commit() {
+			Ok(()) => self.storage.state_db.apply_pending(),
+			e @ Err(_) => {
+				self.storage.state_db.revert_pending();
+				return e;
 			}
-			self.storage.db.write(transaction).map_err(db_err)?;
-			self.blockchain.update_meta(hash, header.number().clone(), false, true);
-			Ok(())
-		} else {
-			Err(client::error::ErrorKind::UnknownBlock(format!("Cannot finalize block {:?}", block)).into())
 		}
+		Ok(())
 	}
 
 	fn changes_trie_storage(&self) -> Option<&Self::ChangesTrieStorage> {

core/client/db/src/lib.rs

@@ -6,7 +6,7 @@ if cargo --version | grep -q "nightly"; then
 else
 	CARGO_CMD="cargo +nightly"
 fi
-$CARGO_CMD build --target=wasm32-unknown-unknown --release
+RUSTFLAGS="-C link-arg=--export-table" $CARGO_CMD build --target=wasm32-unknown-unknown --release
 for i in test
 do
 	wasm-gc target/wasm32-unknown-unknown/release/runtime_$i.wasm target/wasm32-unknown-unknown/release/runtime_$i.compact.wasm