diff --git a/chain/src/chain.rs b/chain/src/chain.rs index aabadc32..13325fad 100644 --- a/chain/src/chain.rs +++ b/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; + } } } diff --git a/chain/src/linked_list.rs b/chain/src/linked_list.rs index 289ffd10..11b0a114 100644 --- a/chain/src/linked_list.rs +++ b/chain/src/linked_list.rs @@ -409,13 +409,17 @@ impl PruneableListIndex for MultiIndex { 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: {}", diff --git a/chain/src/store.rs b/chain/src/store.rs index 3a3df575..e5ed2c60 100644 --- a/chain/src/store.rs +++ b/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, CommitPos)>, Error> { + pub fn output_pos_iter( + &self, + ) -> Result, 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 + 'a, Error> { + pub fn blocks_iter(&self) -> Result> + '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, Vec)>, Error> { + pub fn blocks_raw_iter( + &self, + ) -> Result, Vec), Error>>, Error> { self.db .iter(Some(BLOCK_PREFIX), |k, v| Ok((k.to_vec(), v.to_vec()))) } diff --git a/chain/src/txhashset/txhashset.rs b/chain/src/txhashset/txhashset.rs index c9c55a70..8155a547 100644 --- a/chain/src/txhashset/txhashset.rs +++ b/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", diff --git a/p2p/src/peers.rs b/p2p/src/peers.rs index 790622af..e62a2724 100644 --- a/p2p/src/peers.rs +++ b/p2p/src/peers.rs @@ -281,7 +281,10 @@ impl Peers { pub fn all_peer_data(&self) -> Vec { 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![] diff --git a/p2p/src/store.rs b/p2p/src/store.rs index e0f32e9f..fe404e05 100644 --- a/p2p/src/store.rs +++ b/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, Error> { + pub fn peers_iter(&self) -> Result>, 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, 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, Error> { - let peers: Vec = self.peers_iter()?.collect(); + let peers: Vec = 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) + } } } diff --git a/store/src/lib.rs b/store/src/lib.rs index 5814bf6c..b2408ca5 100644 --- a/store/src/lib.rs +++ b/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>(prefix: u8, k: K) -> Vec { + 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>(prefix: u8, k: K, val: u64) -> Vec { + 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::(val).unwrap(); + res +} +/// Build a db key from a prefix and a numeric identifier. +pub fn u64_to_key(prefix: u8, val: u64) -> Vec { + let mut res = Vec::with_capacity(10); + res.push(prefix); + res.push(SEP); + res.write_u64::(val).unwrap(); + res +} + pub use crate::lmdb::*; use std::ffi::OsStr; diff --git a/store/src/lmdb.rs b/store/src/lmdb.rs index 6d0b9405..ee4c977f 100644 --- a/store/src/lmdb.rs +++ b/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>> = OnceLock::new(); /// State of active database environment. struct EnvState { env: Env, - 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) -> &Database { + fn get_db(&self, db_key: Option) -> Result<&Database, 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, { - 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, 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, key: &[u8]) -> Result { 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, 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>, read: Arc>, keys: Vec>, - skip: usize, + total_keys: usize, + skip_cur: usize, + skip_total: usize, deserialize: F, #[allow(dead_code)] tx_counter: TxCounter, @@ -733,20 +807,47 @@ impl Iterator for DatabaseIterator where F: Fn(&[u8], &[u8]) -> Result, { - type Item = T; + type Item = Result; fn next(&mut self) -> Option { - 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::>>() + } 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 { - 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::>>() + 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::>>() + } 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(), diff --git a/store/tests/lmdb.rs b/store/tests/lmdb.rs index f8c68386..797614db 100644 --- a/store/tests/lmdb.rs +++ b/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, Env), 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 = 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::>(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::(test_key_64_2) + .unwrap(); + assert!(store.exists(Some(test_prefix_2), &key)?); + let data = store.get_ser::>(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::>(None, &test_key_3, None)?; + assert_eq!(data, Some(test_data_3.to_vec())); + } + + clean_output_dir(test_dir); Ok(()) }