alter DB schema to fix lookup race (#1135)

substrate/core/client/db/src/cache/list_storage.rs

@@ -86,7 +86,7 @@ pub struct DbColumns {
 	/// Column holding cache meta.
 	pub meta: Option<u32>,
 	/// Column holding the mapping of { block number => block hash } for blocks of the best chain.
-	pub hash_lookup: Option<u32>,
+	pub key_lookup: Option<u32>,
 	/// Column holding the mapping of { block hash => block header }.
 	pub header: Option<u32>,
 	/// Column holding cache entries.
@@ -126,12 +126,12 @@ impl DbStorage {
 
 impl<Block: BlockT, T: CacheItemT> Storage<Block, T> for DbStorage {
 	fn read_id(&self, at: NumberFor<Block>) -> ClientResult<Option<Block::Hash>> {
-		utils::read_header::<Block>(&*self.db, self.columns.hash_lookup, self.columns.header, BlockId::Number(at))
+		utils::read_header::<Block>(&*self.db, self.columns.key_lookup, self.columns.header, BlockId::Number(at))
 			.map(|maybe_header| maybe_header.map(|header| header.hash()))
 	}
 
 	fn read_header(&self, at: &Block::Hash) -> ClientResult<Option<Block::Header>> {
-		utils::read_header::<Block>(&*self.db, self.columns.hash_lookup, self.columns.header, BlockId::Hash(*at))
+		utils::read_header::<Block>(&*self.db, self.columns.key_lookup, self.columns.header, BlockId::Hash(*at))
 	}
 
 	fn read_meta(&self) -> ClientResult<Metadata<Block>> {
@@ -197,7 +197,7 @@ impl<'a, Block: BlockT, T: CacheItemT> StorageTransaction<Block, T> for DbStorag
 mod meta {
 	use super::*;
 
-	/// Convert cache name into cache metadata key. 
+	/// Convert cache name into cache metadata key.
 	pub fn key(name: &[u8]) -> Vec<u8> {
 		let mut key_name = meta_keys::CACHE_META_PREFIX.to_vec();
 		key_name.extend_from_slice(name);

substrate/core/client/db/src/cache/mod.rs

@@ -72,7 +72,7 @@ impl<Block: BlockT> DbCache<Block> {
 	/// Create new cache.
 	pub fn new(
 		db: Arc<KeyValueDB>,
-		hash_lookup_column: Option<u32>,
+		key_lookup_column: Option<u32>,
 		header_column: Option<u32>,
 		authorities_column: Option<u32>,
 		best_finalized_block: ComplexBlockId<Block>,
@@ -82,7 +82,7 @@ impl<Block: BlockT> DbCache<Block> {
 				self::list_storage::DbStorage::new(b"auth".to_vec(), db,
 					self::list_storage::DbColumns {
 						meta: COLUMN_META,
-						hash_lookup: hash_lookup_column,
+						key_lookup: key_lookup_column,
 						header: header_column,
 						cache: authorities_column,
 					},
@@ -188,7 +188,7 @@ impl<Block: BlockT> BlockchainCache<Block> for DbCacheSync<Block> {
 			BlockId::Hash(hash) => {
 				let header = utils::read_header::<Block>(
 					&**db,
-					columns.hash_lookup,
+					columns.key_lookup,
 					columns.header,
 					BlockId::Hash(hash.clone())).ok()??;
 				ComplexBlockId::new(hash, *header.number())
@@ -196,7 +196,7 @@ impl<Block: BlockT> BlockchainCache<Block> for DbCacheSync<Block> {
 			BlockId::Number(number) => {
 				let hash = utils::read_header::<Block>(
 					&**db,
-					columns.hash_lookup,
+					columns.key_lookup,
 					columns.header,
 					BlockId::Number(number.clone())).ok()??.hash();
 				ComplexBlockId::new(hash, number)

substrate/core/client/db/src/lib.rs

@@ -115,8 +115,8 @@ mod columns {
 	pub const META: Option<u32> = ::utils::COLUMN_META;
 	pub const STATE: Option<u32> = Some(1);
 	pub const STATE_META: Option<u32> = Some(2);
-	/// maps hashes to lookup keys
-	pub const HASH_LOOKUP: Option<u32> = Some(3);
+	/// maps hashes to lookup keys and numbers to canon hashes.
+	pub const KEY_LOOKUP: Option<u32> = Some(3);
 	pub const HEADER: Option<u32> = Some(4);
 	pub const BODY: Option<u32> = Some(5);
 	pub const JUSTIFICATION: Option<u32> = Some(6);
@@ -187,7 +187,7 @@ impl<Block: BlockT> BlockchainDb<Block> {
 
 impl<Block: BlockT> client::blockchain::HeaderBackend<Block> for BlockchainDb<Block> {
 	fn header(&self, id: BlockId<Block>) -> Result<Option<Block::Header>, client::error::Error> {
-		::utils::read_header(&*self.db, columns::HASH_LOOKUP, columns::HEADER, id)
+		::utils::read_header(&*self.db, columns::KEY_LOOKUP, columns::HEADER, id)
 	}
 
 	fn info(&self) -> Result<client::blockchain::Info<Block>, client::error::Error> {
@@ -205,7 +205,7 @@ impl<Block: BlockT> client::blockchain::HeaderBackend<Block> for BlockchainDb<Bl
 		let exists = match id {
 			BlockId::Hash(_) => read_db(
 				&*self.db,
-				columns::HASH_LOOKUP,
+				columns::KEY_LOOKUP,
 				columns::HEADER,
 				id
 			)?.is_some(),
@@ -218,7 +218,7 @@ impl<Block: BlockT> client::blockchain::HeaderBackend<Block> for BlockchainDb<Bl
 	}
 
 	fn number(&self, hash: Block::Hash) -> Result<Option<NumberFor<Block>>, client::error::Error> {
-		if let Some(lookup_key) = block_id_to_lookup_key::<Block>(&*self.db, columns::HASH_LOOKUP, BlockId::Hash(hash))? {
+		if let Some(lookup_key) = block_id_to_lookup_key::<Block>(&*self.db, columns::KEY_LOOKUP, BlockId::Hash(hash))? {
 			let number = utils::lookup_key_to_number(&lookup_key)?;
 			Ok(Some(number))
 		} else {
@@ -236,7 +236,7 @@ impl<Block: BlockT> client::blockchain::HeaderBackend<Block> for BlockchainDb<Bl
 
 impl<Block: BlockT> client::blockchain::Backend<Block> for BlockchainDb<Block> {
 	fn body(&self, id: BlockId<Block>) -> Result<Option<Vec<Block::Extrinsic>>, client::error::Error> {
-		match read_db(&*self.db, columns::HASH_LOOKUP, columns::BODY, id)? {
+		match read_db(&*self.db, columns::KEY_LOOKUP, columns::BODY, id)? {
 			Some(body) => match Decode::decode(&mut &body[..]) {
 				Some(body) => Ok(Some(body)),
 				None => return Err(client::error::ErrorKind::Backend("Error decoding body".into()).into()),
@@ -246,7 +246,7 @@ impl<Block: BlockT> client::blockchain::Backend<Block> for BlockchainDb<Block> {
 	}
 
 	fn justification(&self, id: BlockId<Block>) -> Result<Option<Justification>, client::error::Error> {
-		match read_db(&*self.db, columns::HASH_LOOKUP, columns::JUSTIFICATION, id)? {
+		match read_db(&*self.db, columns::KEY_LOOKUP, columns::JUSTIFICATION, id)? {
 			Some(justification) => match Decode::decode(&mut &justification[..]) {
 				Some(justification) => Ok(Some(justification)),
 				None => return Err(client::error::ErrorKind::Backend("Error decoding justification".into()).into()),
@@ -460,7 +460,7 @@ impl<Block: BlockT> state_machine::ChangesTrieRootsStorage<Blake2Hasher> for DbC
 			let mut current_num = anchor.number;
 			let mut current_hash: Block::Hash = convert_hash(&anchor.hash);
 			let maybe_anchor_header: Block::Header = ::utils::require_header::<Block>(
-				&*self.db, columns::HASH_LOOKUP, columns::HEADER, BlockId::Number(As::sa(current_num))
+				&*self.db, columns::KEY_LOOKUP, columns::HEADER, BlockId::Number(As::sa(current_num))
 			).map_err(|e| e.to_string())?;
 			if maybe_anchor_header.hash() == current_hash {
 				// if anchor is canonicalized, then the block is also canonicalized
@@ -471,7 +471,7 @@ impl<Block: BlockT> state_machine::ChangesTrieRootsStorage<Blake2Hasher> for DbC
 				// back from the anchor to the block with given number
 				while current_num != block {
 					let current_header: Block::Header = ::utils::require_header::<Block>(
-						&*self.db, columns::HASH_LOOKUP, columns::HEADER, BlockId::Hash(current_hash)
+						&*self.db, columns::KEY_LOOKUP, columns::HEADER, BlockId::Hash(current_hash)
 					).map_err(|e| e.to_string())?;
 
 					current_hash = *current_header.parent_hash();
@@ -482,7 +482,7 @@ impl<Block: BlockT> state_machine::ChangesTrieRootsStorage<Blake2Hasher> for DbC
 			}
 		};
 
-		Ok(::utils::require_header::<Block>(&*self.db, columns::HASH_LOOKUP, columns::HEADER, block_id)
+		Ok(::utils::require_header::<Block>(&*self.db, columns::KEY_LOOKUP, columns::HEADER, block_id)
 			.map_err(|e| e.to_string())?
 			.digest().log(DigestItem::as_changes_trie_root)
 			.map(|root| H256::from_slice(root.as_ref())))
@@ -609,7 +609,7 @@ impl<Block: BlockT> Backend<Block> {
 				).into())
 			}
 
-			let lookup_key = ::utils::number_to_lookup_key(f_num);
+			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);
@@ -670,13 +670,8 @@ impl<Block> client::backend::Backend<Block, Blake2Hasher> for Backend<Block> whe
 			let parent_hash = *pending_block.header.parent_hash();
 			let number = pending_block.header.number().clone();
 
-			// blocks in longest chain are keyed by number
-			let lookup_key = if pending_block.leaf_state.is_best() {
-				::utils::number_to_lookup_key(number).to_vec()
-			} else {
-				// other blocks are keyed by number + hash
-				::utils::number_and_hash_to_lookup_key(number, hash)
-			};
+			// blocks are keyed by number + hash.
+			let lookup_key = ::utils::number_and_hash_to_lookup_key(number, hash);
 
 			if pending_block.leaf_state.is_best() {
 				let meta = self.blockchain.meta.read();
@@ -689,7 +684,8 @@ impl<Block> client::backend::Backend<Block, Blake2Hasher> for Backend<Block> whe
 						BlockId::Hash(parent_hash),
 					)?;
 
-					// uncanonicalize
+					// uncanonicalize: check safety violations and ensure the numbers no longer
+					// point to these block hashes in the key mapping.
 					for retracted in tree_route.retracted() {
 						if retracted.hash == meta.finalized_hash {
 							warn!("Potential safety failure: reverting finalized block {:?}",
@@ -698,76 +694,40 @@ impl<Block> client::backend::Backend<Block, Blake2Hasher> for Backend<Block> whe
 							return Err(::client::error::ErrorKind::NotInFinalizedChain.into());
 						}
 
-						let prev_lookup_key = ::utils::number_to_lookup_key(retracted.number);
-						let new_lookup_key = ::utils::number_and_hash_to_lookup_key(retracted.number, retracted.hash);
-
-						// change mapping from `number -> header`
-						// to `number + hash -> header`
-						let retracted_header = if let Some(header) = ::client::blockchain::HeaderBackend::<Block>::header(&self.blockchain, BlockId::Number(retracted.number))? {
-							header
-						} else {
-							return Err(client::error::ErrorKind::UnknownBlock(format!("retracted {:?}", retracted)).into());
-						};
-						transaction.delete(columns::HEADER, &prev_lookup_key);
-						transaction.put(columns::HEADER, &new_lookup_key, &retracted_header.encode());
-
-						// if body is stored
-						// change mapping from `number -> body`
-						// to `number + hash -> body`
-						if let Some(retracted_body) = ::client::blockchain::Backend::<Block>::body(&self.blockchain, BlockId::Number(retracted.number))? {
-							transaction.delete(columns::BODY, &prev_lookup_key);
-							transaction.put(columns::BODY, &new_lookup_key, &retracted_body.encode());
-						}
-
-						// if justification is stored
-						// change mapping from `number -> justification`
-						// to `number + hash -> justification`
-						if let Some(retracted_justification) = ::client::blockchain::Backend::<Block>::justification(&self.blockchain, BlockId::Number(retracted.number))? {
-							transaction.delete(columns::JUSTIFICATION, &prev_lookup_key);
-							transaction.put(columns::JUSTIFICATION, &new_lookup_key, &retracted_justification.encode());
-						}
-
-						transaction.put(columns::HASH_LOOKUP, retracted.hash.as_ref(), &new_lookup_key);
+						::utils::remove_number_to_key_mapping(
+							&mut transaction,
+							columns::KEY_LOOKUP,
+							retracted.number
+						);
 					}
 
-					// canonicalize
+					// canonicalize: set the number lookup to map to this block's hash.
 					for enacted in tree_route.enacted() {
-						let prev_lookup_key = ::utils::number_and_hash_to_lookup_key(enacted.number, enacted.hash);
-						let new_lookup_key = ::utils::number_to_lookup_key(enacted.number);
-
-						// change mapping from `number + hash -> header`
-						// to `number -> header`
-						let enacted_header = if let Some(header) = ::client::blockchain::HeaderBackend::<Block>::header(&self.blockchain, BlockId::Number(enacted.number))? {
-							header
-						} else {
-							return Err(client::error::ErrorKind::UnknownBlock(format!("enacted {:?}", enacted)).into());
-						};
-						transaction.delete(columns::HEADER, &prev_lookup_key);
-						transaction.put(columns::HEADER, &new_lookup_key, &enacted_header.encode());
-
-						// if body is stored
-						// change mapping from `number + hash -> body`
-						// to `number -> body`
-						if let Some(enacted_body) = ::client::blockchain::Backend::<Block>::body(&self.blockchain, BlockId::Number(enacted.number))? {
-							transaction.delete(columns::BODY, &prev_lookup_key);
-							transaction.put(columns::BODY, &new_lookup_key, &enacted_body.encode());
-						}
-
-						// if justification is stored
-						// change mapping from `number -> justification`
-						// to `number + hash -> justification`
-						if let Some(enacted_justification) = ::client::blockchain::Backend::<Block>::justification(&self.blockchain, BlockId::Number(enacted.number))? {
-							transaction.delete(columns::JUSTIFICATION, &prev_lookup_key);
-							transaction.put(columns::JUSTIFICATION, &new_lookup_key, &enacted_justification.encode());
-						}
-
-						transaction.put(columns::HASH_LOOKUP, enacted.hash.as_ref(), &new_lookup_key);
+						::utils::insert_number_to_key_mapping(
+							&mut transaction,
+							columns::KEY_LOOKUP,
+							enacted.number,
+							enacted.hash
+						);
 					}
 				}
 
 				transaction.put(columns::META, meta_keys::BEST_BLOCK, &lookup_key);
+				::utils::insert_number_to_key_mapping(
+					&mut transaction,
+					columns::KEY_LOOKUP,
+					number,
+					hash,
+				);
 			}
 
+			::utils::insert_hash_to_key_mapping(
+				&mut transaction,
+				columns::KEY_LOOKUP,
+				number,
+				hash,
+			);
+
 			transaction.put(columns::HEADER, &lookup_key, &pending_block.header.encode());
 			if let Some(body) = pending_block.body {
 				transaction.put(columns::BODY, &lookup_key, &body.encode());
@@ -776,8 +736,6 @@ impl<Block> client::backend::Backend<Block, Blake2Hasher> for Backend<Block> whe
 				transaction.put(columns::JUSTIFICATION, &lookup_key, &justification.encode());
 			}
 
-			transaction.put(columns::HASH_LOOKUP, hash.as_ref(), &lookup_key);
-
 			if number.is_zero() {
 				transaction.put(columns::META, meta_keys::FINALIZED_BLOCK, &lookup_key);
 				transaction.put(columns::META, meta_keys::GENESIS_HASH, hash.as_ref());
@@ -876,9 +834,9 @@ impl<Block> client::backend::Backend<Block, Blake2Hasher> for Backend<Block> whe
 						|| client::error::ErrorKind::UnknownBlock(
 							format!("Error reverting to {}. Block header not found.", best)))?;
 
-					let lookup_key = ::utils::number_to_lookup_key(header.number().clone());
+					let lookup_key = ::utils::number_and_hash_to_lookup_key(header.number().clone(), header.hash().clone());
 					transaction.put(columns::META, meta_keys::BEST_BLOCK, &lookup_key);
-					transaction.delete(columns::HASH_LOOKUP, header.hash().as_ref());
+					transaction.delete(columns::KEY_LOOKUP, header.hash().as_ref());
 					self.storage.db.write(transaction).map_err(db_err)?;
 					self.blockchain.update_meta(header.hash().clone(), best.clone(), true, false);
 					self.blockchain.leaves.write().revert(header.hash().clone(), header.number().clone(), header.parent_hash().clone());

substrate/core/client/db/src/light.rs

@@ -33,13 +33,13 @@ use runtime_primitives::generic::BlockId;
 use runtime_primitives::traits::{Block as BlockT, Header as HeaderT,
 	Zero, One, As, NumberFor, Digest, DigestItem};
 use cache::{DbCacheSync, DbCache, ComplexBlockId};
-use utils::{meta_keys, Meta, db_err, number_to_lookup_key, open_database,
+use utils::{meta_keys, Meta, db_err, open_database,
 	read_db, block_id_to_lookup_key, read_meta};
 use DatabaseSettings;
 
 pub(crate) mod columns {
 	pub const META: Option<u32> = ::utils::COLUMN_META;
-	pub const HASH_LOOKUP: Option<u32> = Some(1);
+	pub const KEY_LOOKUP: Option<u32> = Some(1);
 	pub const HEADER: Option<u32> = Some(2);
 	pub const CACHE: Option<u32> = Some(3);
 	pub const CHT: Option<u32> = Some(4);
@@ -93,7 +93,7 @@ impl<Block> LightStorage<Block>
 		let leaves = LeafSet::read_from_db(&*db, columns::META, meta_keys::LEAF_PREFIX)?;
 		let cache = DbCache::new(
 			db.clone(),
-			columns::HASH_LOOKUP,
+			columns::KEY_LOOKUP,
 			columns::HEADER,
 			columns::CACHE,
 			ComplexBlockId::new(meta.finalized_hash, meta.finalized_number),
@@ -143,7 +143,7 @@ impl<Block> BlockchainHeaderBackend<Block> for LightStorage<Block>
 		Block: BlockT,
 {
 	fn header(&self, id: BlockId<Block>) -> ClientResult<Option<Block::Header>> {
-		::utils::read_header(&*self.db, columns::HASH_LOOKUP, columns::HEADER, id)
+		::utils::read_header(&*self.db, columns::KEY_LOOKUP, columns::HEADER, id)
 	}
 
 	fn info(&self) -> ClientResult<BlockchainInfo<Block>> {
@@ -161,7 +161,7 @@ impl<Block> BlockchainHeaderBackend<Block> for LightStorage<Block>
 		let exists = match id {
 			BlockId::Hash(_) => read_db(
 				&*self.db,
-				columns::HASH_LOOKUP,
+				columns::KEY_LOOKUP,
 				columns::HEADER,
 				id
 			)?.is_some(),
@@ -174,7 +174,7 @@ impl<Block> BlockchainHeaderBackend<Block> for LightStorage<Block>
 	}
 
 	fn number(&self, hash: Block::Hash) -> ClientResult<Option<NumberFor<Block>>> {
-		if let Some(lookup_key) = block_id_to_lookup_key::<Block>(&*self.db, columns::HASH_LOOKUP, BlockId::Hash(hash))? {
+		if let Some(lookup_key) = block_id_to_lookup_key::<Block>(&*self.db, columns::KEY_LOOKUP, BlockId::Hash(hash))? {
 			let number = ::utils::lookup_key_to_number(&lookup_key)?;
 			Ok(Some(number))
 		} else {
@@ -211,7 +211,7 @@ impl<Block: BlockT> LightStorage<Block> {
 			).into())
 		}
 
-		let lookup_key = ::utils::number_to_lookup_key(header.number().clone());
+		let lookup_key = ::utils::number_and_hash_to_lookup_key(header.number().clone(), hash);
 		transaction.put(columns::META, meta_keys::FINALIZED_BLOCK, &lookup_key);
 
 		// build new CHT(s) if required
@@ -249,9 +249,14 @@ impl<Block: BlockT> LightStorage<Block> {
 
 			while prune_block <= new_cht_end {
 				if let Some(hash) = self.hash(prune_block)? {
-					let lookup_key = block_id_to_lookup_key::<Block>(&*self.db, columns::HASH_LOOKUP, BlockId::Number(prune_block))?
+					let lookup_key = block_id_to_lookup_key::<Block>(&*self.db, columns::KEY_LOOKUP, BlockId::Number(prune_block))?
 						.expect("retrieved hash for `prune_block` right above. therefore retrieving lookup key must succeed. q.e.d.");
-					transaction.delete(columns::HASH_LOOKUP, hash.as_ref());
+					::utils::remove_key_mappings(
+						transaction,
+						columns::KEY_LOOKUP,
+						prune_block,
+						hash
+					);
 					transaction.delete(columns::HEADER, &lookup_key);
 				}
 				prune_block += One::one();
@@ -294,14 +299,6 @@ impl<Block> LightBlockchainStorage<Block> for LightStorage<Block>
 		let number = *header.number();
 		let parent_hash = *header.parent_hash();
 
-		// blocks in longest chain are keyed by number
-		let lookup_key = if leaf_state.is_best() {
-			::utils::number_to_lookup_key(number).to_vec()
-		} else {
-		// other blocks are keyed by number + hash
-			::utils::number_and_hash_to_lookup_key(number, hash)
-		};
-
 		for (key, maybe_val) in aux_ops {
 			match maybe_val {
 				Some(val) => transaction.put_vec(columns::AUX, &key, val),
@@ -309,6 +306,9 @@ impl<Block> LightBlockchainStorage<Block> for LightStorage<Block>
 			}
 		}
 
+		// blocks are keyed by number + hash.
+		let lookup_key = ::utils::number_and_hash_to_lookup_key(number, hash);
+
 		if leaf_state.is_best() {
 			// handle reorg.
 			{
@@ -328,46 +328,40 @@ impl<Block> LightBlockchainStorage<Block> for LightStorage<Block>
 								(&retracted.number, &retracted.hash));
 						}
 
-						let prev_lookup_key = ::utils::number_to_lookup_key(retracted.number);
-						let new_lookup_key = ::utils::number_and_hash_to_lookup_key(retracted.number, retracted.hash);
-
-						// change mapping from `number -> header`
-						// to `number + hash -> header`
-						let retracted_header = if let Some(header) = self.header(BlockId::Number(retracted.number))? {
-							header
-						} else {
-							return Err(::client::error::ErrorKind::UnknownBlock(format!("retracted {:?}", retracted)).into());
-						};
-						transaction.delete(columns::HEADER, &prev_lookup_key);
-						transaction.put(columns::HEADER, &new_lookup_key, &retracted_header.encode());
-
-						transaction.put(columns::HASH_LOOKUP, retracted.hash.as_ref(), &new_lookup_key);
+						::utils::remove_number_to_key_mapping(
+							&mut transaction,
+							columns::KEY_LOOKUP,
+							retracted.number
+						);
 					}
 
 					for enacted in tree_route.enacted() {
-						let prev_lookup_key = ::utils::number_and_hash_to_lookup_key(enacted.number, enacted.hash);
-						let new_lookup_key = ::utils::number_to_lookup_key(enacted.number);
-
-						// change mapping from `number + hash -> header`
-						// to `number -> header`
-						let enacted_header = if let Some(header) = self.header(BlockId::Number(enacted.number))? {
-							header
-						} else {
-							return Err(::client::error::ErrorKind::UnknownBlock(format!("enacted {:?}", enacted)).into());
-						};
-						transaction.delete(columns::HEADER, &prev_lookup_key);
-						transaction.put(columns::HEADER, &new_lookup_key, &enacted_header.encode());
-
-						transaction.put(columns::HASH_LOOKUP, enacted.hash.as_ref(), &new_lookup_key);
+						::utils::insert_number_to_key_mapping(
+							&mut transaction,
+							columns::KEY_LOOKUP,
+							enacted.number,
+							enacted.hash
+						);
 					}
 				}
 			}
 
 			transaction.put(columns::META, meta_keys::BEST_BLOCK, &lookup_key);
+			::utils::insert_number_to_key_mapping(
+				&mut transaction,
+				columns::KEY_LOOKUP,
+				number,
+				hash,
+			);
 		}
 
+		::utils::insert_hash_to_key_mapping(
+			&mut transaction,
+			columns::KEY_LOOKUP,
+			number,
+			hash,
+		);
 		transaction.put(columns::HEADER, &lookup_key, &header.encode());
-		transaction.put(columns::HASH_LOOKUP, hash.as_ref(), &lookup_key);
 
 		if number.is_zero() {
 			transaction.put(columns::META, meta_keys::FINALIZED_BLOCK, &lookup_key);
@@ -462,7 +456,7 @@ impl<Block> LightBlockchainStorage<Block> for LightStorage<Block>
 /// Build the key for inserting header-CHT at given block.
 fn cht_key<N: As<u64>>(cht_type: u8, block: N) -> [u8; 5] {
 	let mut key = [cht_type; 5];
-	key[1..].copy_from_slice(&number_to_lookup_key(block));
+	key[1..].copy_from_slice(&::utils::number_index_key(block));
 	key
 }
 
@@ -585,11 +579,11 @@ pub(crate) mod tests {
 
 		let genesis_hash = insert_block(&db, None, || default_header(&Default::default(), 0));
 		assert_eq!(db.db.iter(columns::HEADER).count(), 1);
-		assert_eq!(db.db.iter(columns::HASH_LOOKUP).count(), 1);
+		assert_eq!(db.db.iter(columns::KEY_LOOKUP).count(), 2);
 
 		let _ = insert_block(&db, None, || default_header(&genesis_hash, 1));
 		assert_eq!(db.db.iter(columns::HEADER).count(), 2);
-		assert_eq!(db.db.iter(columns::HASH_LOOKUP).count(), 2);
+		assert_eq!(db.db.iter(columns::KEY_LOOKUP).count(), 4);
 	}
 
 	#[test]
@@ -631,9 +625,9 @@ pub(crate) mod tests {
 		// when headers are created without changes tries roots
 		let db = insert_headers(default_header);
 		assert_eq!(db.db.iter(columns::HEADER).count(), (1 + cht::SIZE + 1) as usize);
-		assert_eq!(db.db.iter(columns::HASH_LOOKUP).count(), (1 + cht::SIZE + 1) as usize);
+		assert_eq!(db.db.iter(columns::KEY_LOOKUP).count(), (2 * (1 + cht::SIZE + 1)) as usize);
 		assert_eq!(db.db.iter(columns::CHT).count(), 1);
-		assert!((0..cht::SIZE).all(|i| db.db.get(columns::HEADER, &number_to_lookup_key(1 + i)).unwrap().is_none()));
+		assert!((0..cht::SIZE).all(|i| db.header(BlockId::Number(1 + i)).unwrap().is_none()));
 		assert!(db.header_cht_root(cht::SIZE, cht::SIZE / 2).is_ok());
 		assert!(db.header_cht_root(cht::SIZE, cht::SIZE + cht::SIZE / 2).is_err());
 		assert!(db.changes_trie_cht_root(cht::SIZE, cht::SIZE / 2).is_err());
@@ -643,7 +637,7 @@ pub(crate) mod tests {
 		let db = insert_headers(header_with_changes_trie);
 		assert_eq!(db.db.iter(columns::HEADER).count(), (1 + cht::SIZE + 1) as usize);
 		assert_eq!(db.db.iter(columns::CHT).count(), 2);
-		assert!((0..cht::SIZE).all(|i| db.db.get(columns::HEADER, &number_to_lookup_key(1 + i)).unwrap().is_none()));
+		assert!((0..cht::SIZE).all(|i| db.header(BlockId::Number(1 + i)).unwrap().is_none()));
 		assert!(db.header_cht_root(cht::SIZE, cht::SIZE / 2).is_ok());
 		assert!(db.header_cht_root(cht::SIZE, cht::SIZE + cht::SIZE / 2).is_err());
 		assert!(db.changes_trie_cht_root(cht::SIZE, cht::SIZE / 2).is_ok());

substrate/core/client/db/src/utils.rs

@@ -68,11 +68,14 @@ pub struct Meta<N, H> {
 }
 
 /// A block lookup key: used for canonical lookup from block number to hash
-pub type ShortBlockLookupKey = [u8; 4];
+pub type NumberIndexKey = [u8; 4];
 
 /// Convert block number into short lookup key (LE representation) for
 /// blocks that are in the canonical chain.
-pub fn number_to_lookup_key<N>(n: N) -> ShortBlockLookupKey where N: As<u64> {
+///
+/// In the current database schema, this kind of key is only used for
+/// lookups into an index, NOT for storing header data or others.
+pub fn number_index_key<N>(n: N) -> NumberIndexKey where N: As<u64> {
 	let n: u64 = n.as_();
 	assert!(n & 0xffffffff00000000 == 0);
 
@@ -90,7 +93,7 @@ pub fn number_and_hash_to_lookup_key<N, H>(number: N, hash: H) -> Vec<u8> where
 	N: As<u64>,
 	H: AsRef<[u8]>
 {
-	let mut lookup_key = number_to_lookup_key(number).to_vec();
+	let mut lookup_key = number_index_key(number).to_vec();
 	lookup_key.extend_from_slice(hash.as_ref());
 	lookup_key
 }
@@ -107,25 +110,76 @@ pub fn lookup_key_to_number<N>(key: &[u8]) -> client::error::Result<N> where N:
 		| (key[3] as u64)).map(As::sa)
 }
 
+/// Delete number to hash mapping in DB transaction.
+pub fn remove_number_to_key_mapping<N: As<u64>>(
+	transaction: &mut DBTransaction,
+	key_lookup_col: Option<u32>,
+	number: N,
+) {
+	transaction.delete(key_lookup_col, number_index_key(number).as_ref())
+}
+
+/// Remove key mappings.
+pub fn remove_key_mappings<N: As<u64>, H: AsRef<[u8]>>(
+	transaction: &mut DBTransaction,
+	key_lookup_col: Option<u32>,
+	number: N,
+	hash: H,
+) {
+	remove_number_to_key_mapping(transaction, key_lookup_col, number);
+	transaction.delete(key_lookup_col, hash.as_ref());
+}
+
+/// Place a number mapping into the database. This maps number to current perceived
+/// block hash at that position.
+pub fn insert_number_to_key_mapping<N: As<u64> + Clone, H: AsRef<[u8]>>(
+	transaction: &mut DBTransaction,
+	key_lookup_col: Option<u32>,
+	number: N,
+	hash: H,
+) {
+	transaction.put_vec(
+		key_lookup_col,
+		number_index_key(number.clone()).as_ref(),
+		number_and_hash_to_lookup_key(number, hash),
+	)
+}
+
+/// Insert a hash to key mapping in the database.
+pub fn insert_hash_to_key_mapping<N: As<u64>, H: AsRef<[u8]> + Clone>(
+	transaction: &mut DBTransaction,
+	key_lookup_col: Option<u32>,
+	number: N,
+	hash: H,
+) {
+	transaction.put_vec(
+		key_lookup_col,
+		hash.clone().as_ref(),
+		number_and_hash_to_lookup_key(number, hash),
+	)
+}
+
 /// Convert block id to block lookup key.
 /// block lookup key is the DB-key header, block and justification are stored under.
 /// looks up lookup key by hash from DB as necessary.
 pub fn block_id_to_lookup_key<Block>(
 	db: &KeyValueDB,
-	hash_lookup_col: Option<u32>,
+	key_lookup_col: Option<u32>,
 	id: BlockId<Block>
 ) -> Result<Option<Vec<u8>>, client::error::Error> where
 	Block: BlockT,
+	::runtime_primitives::traits::NumberFor<Block>: As<u64>,
 {
-	match id {
-		// numbers are solely looked up in canonical chain
-		BlockId::Number(n) => Ok(Some(number_to_lookup_key(n).to_vec())),
-		BlockId::Hash(h) => db.get(hash_lookup_col, h.as_ref()).map(|v|
-			v.map(|v| { v.into_vec() })
-		).map_err(db_err),
-	}
-}
+	let res = match id {
+		BlockId::Number(n) => db.get(
+			key_lookup_col,
+			number_index_key(n).as_ref(),
+		),
+		BlockId::Hash(h) => db.get(key_lookup_col, h.as_ref()),
+	};
 
+	res.map(|v| v.map(|v| v.into_vec())).map_err(db_err)
+}
 
 /// Maps database error to client error
 pub fn db_err(err: io::Error) -> client::error::Error {