Merge branch 'lmdb_update' into grim

chain/src/chain.rs

@@ -1299,9 +1299,11 @@ impl Chain {
 
 		// Remove old blocks (including short lived fork blocks) which height < tail.height
 		for block in batch.blocks_iter()? {
-			if block.header.height < tail.height {
-				let _ = batch.delete_block(&block.hash());
-				count += 1;
+			if let Ok(block) = block {
+				if block.header.height < tail.height {
+					let _ = batch.delete_block(&block.hash());
+					count += 1;
+				}
 			}
 		}
 

chain/src/linked_list.rs

@@ -409,13 +409,17 @@ impl<T: PosEntry> PruneableListIndex for MultiIndex<T> {
 		let mut entry_count = 0;
 		let list_db_key = Some(self.list_prefix);
 		for key in batch.db.iter(list_db_key, |k, _| Ok(k.to_vec()))? {
-			let _ = batch.delete(list_db_key, &key);
-			list_count += 1;
+			if let Ok(key) = key {
+				let _ = batch.delete(list_db_key, &key);
+				list_count += 1;
+			}
 		}
 		let entry_db_key = Some(self.entry_prefix);
 		for key in batch.db.iter(entry_db_key, |k, _| Ok(k.to_vec()))? {
-			let _ = batch.delete(entry_db_key, &key);
-			entry_count += 1;
+			if let Ok(key) = key {
+				let _ = batch.delete(entry_db_key, &key);
+				entry_count += 1;
+			}
 		}
 		debug!(
 			"clear: lists deleted: {}, entries deleted: {}",

chain/src/store.rs

@@ -339,7 +339,9 @@ impl<'a> Batch<'a> {
 	}
 
 	/// Iterator over the output_pos index.
-	pub fn output_pos_iter(&self) -> Result<impl Iterator<Item = (Vec<u8>, CommitPos)>, Error> {
+	pub fn output_pos_iter(
+		&self,
+	) -> Result<impl Iterator<Item = Result<(Vec<u8>, CommitPos), Error>>, Error> {
 		let protocol_version = self.db.protocol_version();
 		self.db.iter(Some(OUTPUT_POS_PREFIX), move |k, mut v| {
 			ser::deserialize(&mut v, protocol_version, DeserializationMode::default())
@@ -459,7 +461,7 @@ impl<'a> Batch<'a> {
 
 	/// Iterator over all full blocks in the db.
 	/// Uses default db serialization strategy via db protocol version.
-	pub fn blocks_iter(&self) -> Result<impl Iterator<Item = Block> + 'a, Error> {
+	pub fn blocks_iter(&self) -> Result<impl Iterator<Item = Result<Block, Error>> + 'a, Error> {
 		let protocol_version = self.db.protocol_version();
 		self.db.iter(Some(BLOCK_PREFIX), move |_, mut v| {
 			ser::deserialize(&mut v, protocol_version, DeserializationMode::default())
@@ -469,7 +471,9 @@ impl<'a> Batch<'a> {
 
 	/// Iterator over raw data for full blocks in the db.
 	/// Used during block migration (we need flexibility around deserialization).
-	pub fn blocks_raw_iter(&self) -> Result<impl Iterator<Item = (Vec<u8>, Vec<u8>)>, Error> {
+	pub fn blocks_raw_iter(
+		&self,
+	) -> Result<impl Iterator<Item = Result<(Vec<u8>, Vec<u8>), Error>>, Error> {
 		self.db
 			.iter(Some(BLOCK_PREFIX), |k, v| Ok((k.to_vec(), v.to_vec())))
 	}

chain/src/txhashset/txhashset.rs

@@ -644,21 +644,23 @@ impl TxHashSet {
 		// Iterate over the current output_pos index, removing any entries that
 		// do not point to to the expected output.
 		let mut removed_count = 0;
-		for (key, pos1) in batch.output_pos_iter()? {
-			let pos0 = pos1.pos - 1;
-			if let Some(out) = output_pmmr.get_data(pos0) {
-				if let Ok(pos0_via_mmr) = batch.get_output_pos(&out.commitment()) {
-					// If the pos matches and the index key matches the commitment
-					// then keep the entry, other we want to clean it up.
-					if pos0 == pos0_via_mmr
-						&& batch.is_match_output_pos_key(&key, &out.commitment())
-					{
-						continue;
+		for kp in batch.output_pos_iter()? {
+			if let Ok((key, pos1)) = kp {
+				let pos0 = pos1.pos - 1;
+				if let Some(out) = output_pmmr.get_data(pos0) {
+					if let Ok(pos0_via_mmr) = batch.get_output_pos(&out.commitment()) {
+						// If the pos matches and the index key matches the commitment
+						// then keep the entry, other we want to clean it up.
+						if pos0 == pos0_via_mmr
+							&& batch.is_match_output_pos_key(&key, &out.commitment())
+						{
+							continue;
+						}
 					}
 				}
+				batch.delete(Some(store::OUTPUT_POS_PREFIX), &key)?;
+				removed_count += 1;
 			}
-			batch.delete(Some(store::OUTPUT_POS_PREFIX), &key)?;
-			removed_count += 1;
 		}
 		debug!(
 			"init_output_pos_index: removed {} stale index entries",

p2p/src/peers.rs

@@ -281,7 +281,10 @@ impl Peers {
 	pub fn all_peer_data(&self) -> Vec<PeerData> {
 		match self.store.iter_batch() {
 			Ok(batch) => match batch.peers_iter() {
-				Ok(iter) => iter.collect(),
+				Ok(iter) => iter
+					.filter(|p| p.is_ok())
+					.map(|p| p.ok().unwrap())
+					.collect(),
 				Err(e) => {
 					error!("failed to get all peer data: {:?}", e);
 					vec![]

p2p/src/store.rs

@@ -195,7 +195,7 @@ pub struct PeersIterBatch<'a> {
 
 impl<'a> PeersIterBatch<'a> {
 	/// Iterator over all known peers.
-	pub fn peers_iter(&self) -> Result<impl Iterator<Item = PeerData>, Error> {
+	pub fn peers_iter(&self) -> Result<impl Iterator<Item = Result<PeerData, Error>>, Error> {
 		let protocol_version = self.db.protocol_version();
 		self.db.iter(Some(PEER_PREFIX), move |_, mut v| {
 			ser::deserialize(&mut v, protocol_version, DeserializationMode::default())
@@ -212,6 +212,8 @@ impl<'a> PeersIterBatch<'a> {
 	) -> Result<Vec<PeerData>, Error> {
 		let peers = self
 			.peers_iter()?
+			.filter(|p| p.is_ok())
+			.map(|p| p.ok().unwrap())
 			.filter(|p| p.flags == state && p.capabilities.contains(cap))
 			.choose_multiple(&mut thread_rng(), count);
 		Ok(peers)
@@ -220,7 +222,11 @@ impl<'a> PeersIterBatch<'a> {
 	/// List all known peers
 	/// Used for /v1/peers/all api endpoint
 	pub fn all_peers(&self) -> Result<Vec<PeerData>, Error> {
-		let peers: Vec<PeerData> = self.peers_iter()?.collect();
+		let peers: Vec<PeerData> = self
+			.peers_iter()?
+			.filter(|p| p.is_ok())
+			.map(|p| p.ok().unwrap())
+			.collect();
 		Ok(peers)
 	}
 
@@ -232,8 +238,10 @@ impl<'a> PeersIterBatch<'a> {
 		let mut to_remove = vec![];
 
 		for x in self.peers_iter()? {
-			if predicate(&x) {
-				to_remove.push(x)
+			if let Ok(x) = x {
+				if predicate(&x) {
+					to_remove.push(x)
+				}
 			}
 		}
 

store/src/lib.rs

@@ -32,6 +32,39 @@ pub mod pmmr;
 pub mod prune_list;
 pub mod types;
 
+const SEP: u8 = b':';
+
+use byteorder::{BigEndian, WriteBytesExt};
+
+/// Build a db key from a prefix and a byte vector identifier.
+pub fn to_key<K: AsRef<[u8]>>(prefix: u8, k: K) -> Vec<u8> {
+	let k = k.as_ref();
+	let mut res = Vec::with_capacity(k.len() + 2);
+	res.push(prefix);
+	res.push(SEP);
+	res.extend_from_slice(k);
+	res
+}
+
+/// Build a db key from a prefix and a byte vector identifier and numeric identifier
+pub fn to_key_u64<K: AsRef<[u8]>>(prefix: u8, k: K, val: u64) -> Vec<u8> {
+	let k = k.as_ref();
+	let mut res = Vec::with_capacity(k.len() + 10);
+	res.push(prefix);
+	res.push(SEP);
+	res.extend_from_slice(k);
+	res.write_u64::<BigEndian>(val).unwrap();
+	res
+}
+/// Build a db key from a prefix and a numeric identifier.
+pub fn u64_to_key(prefix: u8, val: u64) -> Vec<u8> {
+	let mut res = Vec::with_capacity(10);
+	res.push(prefix);
+	res.push(SEP);
+	res.write_u64::<BigEndian>(val).unwrap();
+	res
+}
+
 pub use crate::lmdb::*;
 
 use std::ffi::OsStr;

store/src/lmdb.rs

@@ -18,6 +18,7 @@ use heed::types::Bytes;
 use heed::{Database, Env, EnvOpenOptions, RoTxn, RwTxn, WithoutTls};
 use std::collections::HashMap;
 use std::path::Path;
+use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
 use std::sync::{Arc, OnceLock};
 use std::time::Duration;
 use std::{fs, thread};
@@ -83,7 +84,7 @@ where
 const DEFAULT_DB_VERSION: ProtocolVersion = ProtocolVersion(3);
 
 /// Default environment.
-const DEFAULT_ENV_NAME: &'static str = "lmdb";
+pub const DEFAULT_ENV_NAME: &'static str = "lmdb";
 /// Default multi-database environment without prefixes.
 const DEFAULT_MULTI_DB_ENV_NAME: &'static str = "multi_lmdb";
 /// Prefix key separator.
@@ -95,10 +96,10 @@ static ENV_MAP: OnceLock<RwLock<HashMap<String, EnvState>>> = OnceLock::new();
 /// State of active database environment.
 struct EnvState {
 	env: Env<WithoutTls>,
-	open_txs_count: u32,
-	resizing: bool,
-	resize_checking: bool,
-	stores_count: u32,
+	open_txs_count: AtomicU32,
+	resizing: AtomicBool,
+	resize_checking: AtomicBool,
+	stores_count: AtomicU32,
 }
 
 /// LMDB-backed store facilitating data access and serialization. All writes
@@ -116,7 +117,16 @@ impl Drop for Store {
 	fn drop(&mut self) {
 		{
 			let mut w_map = ENV_MAP.get().unwrap().write();
-			w_map.get_mut(&self.env_path).unwrap().stores_count -= 1;
+			let stores_count = w_map
+				.get(&self.env_path)
+				.unwrap()
+				.stores_count
+				.load(Ordering::Relaxed);
+			w_map
+				.get_mut(&self.env_path)
+				.unwrap()
+				.stores_count
+				.store(stores_count - 1, Ordering::Relaxed);
 		}
 		let no_stores = {
 			ENV_MAP
@@ -126,6 +136,7 @@ impl Drop for Store {
 				.get(&self.env_path)
 				.unwrap()
 				.stores_count
+				.load(Ordering::Relaxed)
 				== 0
 		};
 		if no_stores {
@@ -193,15 +204,24 @@ impl Store {
 				full_path.clone(),
 				EnvState {
 					env,
-					open_txs_count: 0,
-					resizing: false,
-					resize_checking: false,
-					stores_count: 1,
+					open_txs_count: AtomicU32::new(0),
+					resizing: AtomicBool::new(false),
+					resize_checking: AtomicBool::new(false),
+					stores_count: AtomicU32::new(1),
 				},
 			);
 		} else {
 			let mut w_env_map = env_map.write();
-			w_env_map.get_mut(&full_path).unwrap().stores_count += 1;
+			let stores_count = w_env_map
+				.get(&full_path)
+				.unwrap()
+				.stores_count
+				.load(Ordering::Relaxed);
+			w_env_map
+				.get_mut(&full_path)
+				.unwrap()
+				.stores_count
+				.store(stores_count + 1, Ordering::Relaxed);
 		}
 
 		// Database setup.
@@ -235,7 +255,16 @@ impl Store {
 					Ok(_) => {
 						let _ = fs::remove_dir_all(&migrate_from);
 					}
-					Err(e) => error!("DB {} migration error: {:?}", env_name, e),
+					Err(e) => {
+						error!("DB {} migration error: {:?}", env_name, e);
+						match s.clear() {
+							Ok(_) => {}
+							Err(e) => {
+								error!("Can not clear new DB after unsuccessful migration: {:?}", e)
+							}
+						}
+						return Err(e);
+					}
 				}
 			}
 		}
@@ -249,7 +278,7 @@ impl Store {
 		from_name: Option<&str>,
 		from_path: &Path,
 	) -> Result<(), Error> {
-		debug!("Migrating DB {:?}", from_path);
+		info!("Migrating DB {:?}, please wait...", from_path);
 		let from_env = unsafe {
 			let mut options = EnvOpenOptions::new().read_txn_without_tls();
 			let env_options = options.map_size(self.alloc_chunk_size).max_dbs(24);
@@ -257,10 +286,11 @@ impl Store {
 		};
 		let (resize, new_size) = needs_resize(&from_env, self.alloc_chunk_size);
 		if resize {
+			// We are sure there are no active txs, cause migration is called on database creation.
 			unsafe {
 				from_env.resize(new_size)?;
 				self.env.resize(new_size)?;
-			};
+			}
 		}
 		let db_from = {
 			let mut write = from_env.write_txn()?;
@@ -272,21 +302,23 @@ impl Store {
 		let read_from = from_env.read_txn()?;
 		let mut count = 0;
 		for kv in db_from.iter(&read_from)? {
-			if let Ok((k, v)) = kv {
-				if k.contains(&PREFIX_KEY_SEPARATOR) {
-					let db_name = k.split_at(1).0;
-					let db = self.pre_dbs.get(&db_name[0]).unwrap();
+			let (k, v) = kv?;
+			if k.len() > 1 && k[1] == PREFIX_KEY_SEPARATOR {
+				let db_name = k.split_at(1).0;
+				if let Some(db) = self.pre_dbs.get(&db_name[0]) {
 					let key = k.split_at(2).1;
 					db.put(&mut write_to, key, &v)?;
 					count += 1;
 				} else {
-					self.def_db.put(&mut write_to, k, &v)?;
-					count += 1;
+					error!("Migration: unknown DB key: {}", db_name[0]);
 				}
+			} else {
+				self.def_db.put(&mut write_to, k, &v)?;
+				count += 1;
 			}
 		}
 		write_to.commit()?;
-		debug!("Migrated {} records from DB {:?}", count, from_path);
+		info!("Migrated {} records from {:?}", count, from_path);
 		Ok(())
 	}
 
@@ -299,6 +331,7 @@ impl Store {
 			.get(&self.env_path)
 			.unwrap()
 			.open_txs_count
+			.load(Ordering::Relaxed)
 	}
 
 	/// Check if requirement for environment resize is checking.
@@ -310,6 +343,7 @@ impl Store {
 			.get(&self.env_path)
 			.unwrap()
 			.resize_checking
+			.load(Ordering::Relaxed)
 	}
 
 	/// Set flag if requirement for environment resize is checking.
@@ -320,7 +354,8 @@ impl Store {
 			.write()
 			.get_mut(&self.env_path)
 			.unwrap()
-			.resize_checking = resize_checking;
+			.resize_checking
+			.store(resize_checking, Ordering::Relaxed);
 	}
 
 	/// Wait while database is resizing.
@@ -333,6 +368,7 @@ impl Store {
 				.get(&self.env_path)
 				.unwrap()
 				.resizing
+				.load(Ordering::Relaxed)
 			{
 				break;
 			}
@@ -368,7 +404,8 @@ impl Store {
 							.read()
 							.get(&env_path)
 							.unwrap()
-							.open_txs_count;
+							.open_txs_count
+							.load(Ordering::Relaxed);
 						if txs_count == 0 {
 							debug!("Start resizing DB {}", env_path);
 							ENV_MAP
@@ -377,11 +414,13 @@ impl Store {
 								.write()
 								.get_mut(&env_path)
 								.unwrap()
-								.resizing = true;
+								.resizing
+								.store(true, Ordering::Relaxed);
 							// Wait to make sure there are no more active txs left.
 							thread::sleep(Duration::from_millis(1000));
 							break;
 						}
+						thread::sleep(Duration::from_millis(10));
 					}
 
 					unsafe {
@@ -393,8 +432,8 @@ impl Store {
 
 					let mut w_env_map = ENV_MAP.get().unwrap().write();
 					let env_state = w_env_map.get_mut(&env_path).unwrap();
-					env_state.resizing = false;
-					env_state.resize_checking = false;
+					env_state.resizing.store(false, Ordering::Relaxed);
+					env_state.resize_checking.store(false, Ordering::Relaxed);
 				});
 				return;
 			} else {
@@ -402,30 +441,47 @@ impl Store {
 				let env_state = w_env_map.get_mut(&env_path).unwrap();
 
 				debug!("Start immediate resizing DB {}", env_path);
-				env_state.resizing = true;
+				env_state.resizing.store(true, Ordering::Relaxed);
 				unsafe {
 					match env.resize(new_size) {
 						Ok(_) => debug!("End resizing DB {}", env_path),
 						Err(e) => error!("Resize DB {} error: {:?}", env_path, e),
 					}
 				}
-				env_state.resizing = false;
+				env_state.resizing.store(false, Ordering::Relaxed);
 			}
 		}
 
 		self.set_resize_checking(false);
 	}
 
+	/// Clear all data from database environment.
+	fn clear(&self) -> Result<(), Error> {
+		let mut w = self.env.write_txn()?;
+		self.def_db.clear(&mut w)?;
+		for db in self.pre_dbs.values() {
+			db.clear(&mut w)?;
+		}
+		w.commit()?;
+		Ok(())
+	}
+
 	/// Protocol version for the store.
 	pub fn protocol_version(&self) -> ProtocolVersion {
 		self.version
 	}
 
 	/// Get database from provided key or return default.
-	fn get_db(&self, db_key: Option<u8>) -> &Database<Bytes, Bytes> {
+	fn get_db(&self, db_key: Option<u8>) -> Result<&Database<Bytes, Bytes>, Error> {
 		match db_key {
-			Some(db) => self.pre_dbs.get(&db).unwrap(),
-			None => &self.def_db,
+			Some(db) => {
+				if let Some(db) = self.pre_dbs.get(&db) {
+					Ok(db)
+				} else {
+					Err(Error::OtherErr("db for provided key not found".to_string()))
+				}
+			}
+			None => Ok(&self.def_db),
 		}
 	}
 
@@ -441,7 +497,7 @@ impl Store {
 	where
 		F: Fn(&[u8], &[u8]) -> Result<T, Error>,
 	{
-		let db = self.get_db(db_key);
+		let db = self.get_db(db_key)?;
 		let res: Option<&[u8]> = db.get(read, key)?;
 		match res {
 			None => Ok(None),
@@ -484,10 +540,15 @@ impl Store {
 		let res = {
 			match self.env.read_txn() {
 				Ok(read) => {
-					let db = self.get_db(db_key);
-					let res = db.get(&read, key);
-					match res {
-						Ok(r) => Ok(r.is_some()),
+					let db_res = self.get_db(db_key);
+					match db_res {
+						Ok(db) => {
+							let res = db.get(&read, key);
+							match res {
+								Ok(r) => Ok(r.is_some()),
+								Err(e) => Err(Error::from(e)),
+							}
+						}
 						Err(e) => Err(Error::from(e)),
 					}
 				}
@@ -512,13 +573,19 @@ impl Store {
 		TxCounter::on_change_tx_count(&self.env_path, true);
 		match self.env.clone().static_read_txn() {
 			Ok(read) => {
-				let db = self.get_db(db_key);
-				Ok(DatabaseIterator::new(
-					self,
-					Arc::new(db.clone()),
-					read,
-					deserialize,
-				))
+				let db_res = self.get_db(db_key);
+				match db_res {
+					Ok(db) => Ok(DatabaseIterator::new(
+						self,
+						Arc::new(db.clone()),
+						read,
+						deserialize,
+					)),
+					Err(e) => {
+						TxCounter::on_change_tx_count(&self.env_path, false);
+						Err(Error::from(e))
+					}
+				}
 			}
 			Err(e) => {
 				TxCounter::on_change_tx_count(&self.env_path, false);
@@ -558,10 +625,15 @@ impl TxCounter {
 	fn on_change_tx_count(env_path: &String, inc: bool) {
 		let mut w_env_map = ENV_MAP.get().unwrap().write();
 		let env_state = w_env_map.get_mut(env_path).unwrap();
+		let open_txs_count = env_state.open_txs_count.load(Ordering::Relaxed);
 		if inc {
-			env_state.open_txs_count += 1;
+			env_state
+				.open_txs_count
+				.store(open_txs_count + 1, Ordering::Relaxed);
 		} else {
-			env_state.open_txs_count -= 1;
+			env_state
+				.open_txs_count
+				.store(open_txs_count - 1, Ordering::Relaxed);
 		}
 	}
 }
@@ -589,7 +661,7 @@ impl<'a> Batch<'a> {
 
 	/// Writes a single key/value pair to the provided database key.
 	pub fn put(&mut self, db_key: Option<u8>, key: &[u8], value: &[u8]) -> Result<(), Error> {
-		let db = self.store.get_db(db_key);
+		let db = self.store.get_db(db_key)?;
 		db.put(&mut self.write, key, value)?;
 		Ok(())
 	}
@@ -645,7 +717,7 @@ impl<'a> Batch<'a> {
 	/// This is in the context of the current write transaction.
 	pub fn exists(&self, db_key: Option<u8>, key: &[u8]) -> Result<bool, Error> {
 		let read = self.write.nested_read_txn()?;
-		let db = self.store.get_db(db_key);
+		let db = self.store.get_db(db_key)?;
 		let res = db.get(&read, key)?;
 		Ok(res.is_some())
 	}
@@ -683,7 +755,7 @@ impl<'a> Batch<'a> {
 
 	/// Deletes a key/value pair from the database.
 	pub fn delete(&mut self, db_key: Option<u8>, key: &[u8]) -> Result<(), Error> {
-		let db = self.store.get_db(db_key);
+		let db = self.store.get_db(db_key)?;
 		db.delete(&mut self.write, key)?;
 		Ok(())
 	}
@@ -723,7 +795,9 @@ where
 	db: Arc<Database<Bytes, Bytes>>,
 	read: Arc<RoTxn<'static, WithoutTls>>,
 	keys: Vec<Vec<u8>>,
-	skip: usize,
+	total_keys: usize,
+	skip_cur: usize,
+	skip_total: usize,
 	deserialize: F,
 	#[allow(dead_code)]
 	tx_counter: TxCounter,
@@ -733,20 +807,47 @@ impl<F, T> Iterator for DatabaseIterator<F, T>
 where
 	F: Fn(&[u8], &[u8]) -> Result<T, Error>,
 {
-	type Item = T;
+	type Item = Result<T, Error>;
 
 	fn next(&mut self) -> Option<Self::Item> {
-		if let Some(k) = self.keys.iter().skip(self.skip).next() {
-			let v = self.db.get(&self.read, k).unwrap_or(None);
-			if let Some(v) = v {
-				return match (self.deserialize)(k, v) {
-					Ok(v) => {
-						self.skip += 1;
-						Some(v)
+		if let Some(k) = self.keys.iter().skip(self.skip_cur).next() {
+			match self.db.get(&self.read, k) {
+				Ok(v) => {
+					if let Some(v) = v {
+						return match (self.deserialize)(k, v) {
+							Ok(v) => {
+								self.skip_total += 1;
+								self.skip_cur += 1;
+								Some(Ok(v))
+							}
+							Err(e) => {
+								error!("db iter: error deserializing: {}", e);
+								Some(Err(Error::from(e)))
+							}
+						};
 					}
-					Err(_) => None,
-				};
+				}
+				Err(e) => {
+					return {
+						error!("db iter: error read value: {}", e);
+						Some(Err(Error::from(e)))
+					}
+				}
 			}
+		} else if self.total_keys > self.skip_total {
+			let keys = if let Ok(iter) = self.db.iter(&self.read) {
+				iter.move_between_keys()
+					.skip(self.skip_total)
+					.take(10000)
+					.filter(|kv| kv.is_ok())
+					.map(|kv| kv.unwrap().0.to_vec())
+					.collect::<Vec<Vec<u8>>>()
+			} else {
+				vec![]
+			};
+			self.skip_cur = 0;
+			self.keys = keys;
+			return self.next();
 		}
 		None
 	}
@@ -763,19 +864,28 @@ where
 		read: RoTxn<'static, WithoutTls>,
 		deserialize: F,
 	) -> DatabaseIterator<F, T> {
-		let keys = if let Ok(iter) = db.iter(&read) {
-			iter.move_between_keys()
-				.filter(|kv| kv.is_ok())
-				.map(|kv| kv.unwrap().0.to_vec())
-				.collect::<Vec<Vec<u8>>>()
+		let (keys, total_keys) = if let Ok(iter) = db.iter(&read) {
+			let total = iter.move_between_keys().count();
+			let keys = if let Ok(iter) = db.iter(&read) {
+				iter.move_between_keys()
+					.take(10000)
+					.filter(|kv| kv.is_ok())
+					.map(|kv| kv.unwrap().0.to_vec())
+					.collect::<Vec<Vec<u8>>>()
+			} else {
+				vec![]
+			};
+			(keys, total)
 		} else {
-			vec![]
+			(vec![], 0)
 		};
 		DatabaseIterator {
 			db,
 			read: Arc::new(read),
 			keys,
-			skip: 0,
+			total_keys,
+			skip_cur: 0,
+			skip_total: 0,
 			deserialize,
 			tx_counter: TxCounter {
 				env_path: store.env_path.clone(),

store/tests/lmdb.rs

@@ -16,9 +16,17 @@ use grin_core as core;
 use grin_store as store;
 use grin_util as util;
 
-use crate::core::global;
-use crate::core::ser::{self, Readable, Reader, Writeable, Writer};
+use core::global;
+use core::ser::{self, Readable, Reader, Writeable, Writer};
+use store::{
+	needs_resize, to_key, to_key_u64, Store, ALLOC_CHUNK_SIZE_DEFAULT_TEST, DEFAULT_ENV_NAME,
+};
+
+use byteorder::WriteBytesExt;
+use heed::types::Bytes;
+use heed::{Database, Env, EnvOpenOptions, WithoutTls};
 use std::fs;
+use std::path::Path;
 
 const WRITE_CHUNK_SIZE: usize = 20;
 const TEST_ALLOC_SIZE: usize = store::lmdb::ALLOC_CHUNK_SIZE_DEFAULT / 8 / WRITE_CHUNK_SIZE;
@@ -126,7 +134,7 @@ fn test_iter() -> Result<(), store::Error> {
 
 	// Check we can see the new entry via an iterator after committing the batch.
 	let mut iter = store.iter(Some(prefix), |_, v| Ok(v.to_vec()))?;
-	assert_eq!(iter.next(), Some(value.to_vec()));
+	assert_eq!(iter.next(), Some(Ok(value.to_vec())));
 	assert_eq!(iter.next(), None);
 
 	clean_output_dir(test_dir);
@@ -168,5 +176,110 @@ fn lmdb_allocate() -> Result<(), store::Error> {
 		}
 	}
 
+	clean_output_dir(test_dir);
+	Ok(())
+}
+
+fn create_old_db(
+	test_dir: &str,
+) -> Result<(Database<Bytes, Bytes>, Env<WithoutTls>), store::Error> {
+	let env_name = DEFAULT_ENV_NAME;
+	let alloc_chunk_size = ALLOC_CHUNK_SIZE_DEFAULT_TEST;
+	let full_path = Path::new(test_dir)
+		.join(env_name)
+		.to_str()
+		.unwrap()
+		.to_string();
+	let _ = fs::create_dir_all(&full_path);
+
+	let env = unsafe {
+		let mut options = EnvOpenOptions::new().read_txn_without_tls();
+		let env_options = options.map_size(alloc_chunk_size).max_dbs(1);
+		env_options.open(&full_path)?
+	};
+	let (resize, new_size) = needs_resize(&env, alloc_chunk_size);
+	if resize {
+		unsafe {
+			env.resize(new_size)?;
+		};
+	}
+
+	let mut write = env.write_txn()?;
+	let db: Database<Bytes, Bytes> = env.create_database(&mut write, Some(env_name))?;
+	write.commit()?;
+
+	Ok((db, env))
+}
+
+#[test]
+fn test_migration() -> Result<(), store::Error> {
+	let test_dir = "target/test_migration";
+	setup(test_dir);
+
+	let test_prefix_1 = b'H';
+	let test_key_1 = [0, 1, 2, 4];
+	let test_data_1 = [1, 2, 3, 4];
+
+	let test_prefix_2 = b'G';
+	let test_key_2 = [3, 4, 5, 6];
+	let test_key_64_2 = 65480464;
+	let test_data_2 = [4, 5, 6, 7];
+
+	let test_key_3 = [6, 7, 8, 9];
+	let test_data_3 = [7, 8, 9, 10];
+
+	// Create old db and fill the data.
+	{
+		let (old_db, old_env) = create_old_db(test_dir)?;
+		let mut w = old_env.write_txn()?;
+
+		// Create old format key value.
+		let key_1 = to_key(test_prefix_1, test_key_1);
+		old_db.put(&mut w, key_1.as_slice(), test_data_1.as_slice())?;
+
+		// Create old format 64 key value.
+		let key_2 = to_key_u64(test_prefix_2, test_key_2, test_key_64_2);
+		old_db.put(&mut w, key_2.as_slice(), test_data_2.as_slice())?;
+
+		// Create key value without prefix.
+		old_db.put(&mut w, test_key_3.as_slice(), test_data_3.as_slice())?;
+
+		w.commit()?;
+	}
+
+	// Create new store to migrate data.
+	let store = Store::new(
+		test_dir,
+		None,
+		Some(DEFAULT_ENV_NAME),
+		vec![test_prefix_1, test_prefix_2],
+		None,
+	)?;
+
+	// Check we can see key value.
+	{
+		assert!(store.exists(Some(test_prefix_1), &test_key_1)?);
+		let data = store.get_ser::<Vec<u8>>(Some(test_prefix_1), &test_key_1, None)?;
+		assert_eq!(data, Some(test_data_1.to_vec()));
+	}
+
+	// Check we can see key 64 value.
+	{
+		let mut key = test_key_2.to_vec();
+		key.write_u64::<byteorder::BigEndian>(test_key_64_2)
+			.unwrap();
+		assert!(store.exists(Some(test_prefix_2), &key)?);
+		let data = store.get_ser::<Vec<u8>>(Some(test_prefix_2), &key, None)?;
+		assert_eq!(data, Some(test_data_2.to_vec()));
+	}
+
+	// Check we can see key value without prefix.
+	{
+		assert!(store.exists(None, &test_key_3)?);
+		let data = store.get_ser::<Vec<u8>>(None, &test_key_3, None)?;
+		assert_eq!(data, Some(test_data_3.to_vec()));
+	}
+
+	clean_output_dir(test_dir);
 	Ok(())
 }