Migrate from lmdb-zero to heed (#3825)

* db: migrate from lmdb-zero to heed

* fix: check resizing operation and wait to avoid crash with multiple batches, fix exists check at batch

* build: fix missing deps at Cargo.lock

* lmdb: single environment, migrate existing databases with provided non-default environment name

* fix: revert chunk size to 128mb

* lmdb: ability to use multiple shared environments

* build: remove unused dependency

* fix: resize to have correct multiplier of the system page size

* lmdb: speed up prefix iter by storing keys

* lmdb: default env name

* lmdb: wait db resize before read, reduce timeout before resizing

* lmdb: use static reader for iterator, count existing batches for stable resize

* fix: check batches count on resize waiting

* lmdb: use separate databases instead of prefixes, use default db for values without prefixes, migrate old environment

* fix: pop pos key

* lmdb: count all open transactions to finish before resizing

* lmdb: immediate resize if there are no open transactions

* lmdb: remove env state when there are no more stores

* lmdb: use atomic for resize and resize checking flags

* lmdb: sleep 10ms when waiting all opened txs to be closed

* lmdb: use atomic open txs and stores count

* lmdb: use index to detect separator, ignore unknown db key to not have a panic

* lmdb: store max 10k keys in the iterator

* lmdb: check iter result on getting total

* lmdb: handle errors at iterator

* lmdb: handle an error when db with provided key not found

* lmdb: fix iterate over 10k keys

* lmdb: document migration resize safety

* lmdb: fix iter test

* lmdb: clear new db after unsuccessful migration, handle read error on migration to interrupt process

* store: bring back old key methods to reproduce data migration

* lmdb: return an error on unsuccessful migration

* lmdb: migration test, clean data after allocate test

* lmdb: info migration log

* fix: move iterator before handling an error to allow skip bad value

* lmdb: return an error if removal of old DB file failed after migration

* lmdb: lifetime for iterator, use write transaction at batch iterator

* lmdb: migration progress

* fix: tests

* lmdb: immediately set resizing flag, ignore resizing flag while there are more than 0 opened txs to avoid stuck, optimize tx counter for some operations

* lmdb: key for successful migration

* lmdb: fix put database creation at separate block to avoid lifetime issues when returning an error on migration

Author: ardocrat

Cargo.lock

@@ -43,7 +43,7 @@ use std::collections::HashMap;
 use std::fs::{self, File};
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicUsize, Ordering};
-use std::sync::Arc;
+use std::sync::{mpsc, Arc};
 use std::time::{Duration, Instant};
 
 /// Orphan pool size is limited by MAX_ORPHAN_SIZE
@@ -150,8 +150,8 @@ pub struct Chain {
 	store: Arc<store::ChainStore>,
 	adapter: Arc<dyn ChainAdapter + Send + Sync>,
 	orphans: Arc<OrphanBlockPool>,
-	txhashset: Arc<RwLock<txhashset::TxHashSet>>,
-	header_pmmr: Arc<RwLock<txhashset::PMMRHandle<BlockHeader>>>,
+	txhashset: Arc<RwLock<TxHashSet>>,
+	header_pmmr: Arc<RwLock<PMMRHandle<BlockHeader>>>,
 	pibd_segmenter: Arc<RwLock<Option<Segmenter>>>,
 	pibd_desegmenter: Arc<RwLock<Option<Desegmenter>>>,
 	// POW verification function
@@ -171,8 +171,9 @@ impl Chain {
 		genesis: Block,
 		pow_verifier: fn(&BlockHeader) -> Result<(), pow::Error>,
 		archive_mode: bool,
+		db_migration_prog_tx: Option<mpsc::Sender<i8>>,
 	) -> Result<Chain, Error> {
-		let store = Arc::new(store::ChainStore::new(&db_root)?);
+		let store = Arc::new(store::ChainStore::new(&db_root, db_migration_prog_tx)?);
 
 		// open the txhashset, creating a new one if necessary
 		let mut txhashset = txhashset::TxHashSet::open(db_root.clone(), store.clone(), None)?;
@@ -189,9 +190,9 @@ impl Chain {
 		// Initialize the output_pos index based on UTXO set
 		// and NRD kernel_pos index based recent kernel history.
 		{
-			let batch = store.batch()?;
-			txhashset.init_output_pos_index(&header_pmmr, &batch)?;
-			txhashset.init_recent_kernel_pos_index(&header_pmmr, &batch)?;
+			let mut batch = store.batch()?;
+			txhashset.init_output_pos_index(&header_pmmr, &mut batch)?;
+			txhashset.init_recent_kernel_pos_index(&header_pmmr, &mut batch)?;
 			batch.commit()?;
 		}
 
@@ -275,7 +276,7 @@ impl Chain {
 	pub fn reset_chain_head_to_genesis(&self) -> Result<(), Error> {
 		let mut header_pmmr = self.header_pmmr.write();
 		let mut txhashset = self.txhashset.write();
-		let batch = self.store.batch()?;
+		let mut batch = self.store.batch()?;
 
 		// Change head back to genesis
 		{
@@ -314,7 +315,7 @@ impl Chain {
 
 	/// Reset PIBD head
 	pub fn reset_pibd_head(&self) -> Result<(), Error> {
-		let batch = self.store.batch()?;
+		let mut batch = self.store.batch()?;
 		batch.save_pibd_head(&self.genesis().into())?;
 		Ok(())
 	}
@@ -530,9 +531,9 @@ impl Chain {
 	pub fn new_ctx<'a>(
 		&self,
 		opts: Options,
-		batch: store::Batch<'a>,
-		header_pmmr: &'a mut txhashset::PMMRHandle<BlockHeader>,
-		txhashset: &'a mut txhashset::TxHashSet,
+		batch: Batch<'a>,
+		header_pmmr: &'a mut PMMRHandle<BlockHeader>,
+		txhashset: &'a mut TxHashSet,
 	) -> Result<pipe::BlockContext<'a>, Error> {
 		let denylist = self.denylist.read().clone();
 		Ok(pipe::BlockContext {
@@ -832,7 +833,7 @@ impl Chain {
 		&self,
 		header: &BlockHeader,
 		ext: &mut ExtensionPair,
-		batch: &Batch,
+		batch: &mut Batch,
 	) -> Result<BlockHeader, Error> {
 		let denylist = self.denylist.read().clone();
 		pipe::rewind_and_apply_fork(header, ext, batch, &|header| {
@@ -846,7 +847,7 @@ impl Chain {
 		&self,
 		header: &BlockHeader,
 		ext: &mut HeaderExtension,
-		batch: &Batch,
+		batch: &mut Batch,
 	) -> Result<(), Error> {
 		let denylist = self.denylist.read().clone();
 		pipe::rewind_and_apply_header_fork(header, ext, batch, &|header| {
@@ -1015,7 +1016,7 @@ impl Chain {
 	fn validate_kernel_history(
 		&self,
 		header: &BlockHeader,
-		txhashset: &txhashset::TxHashSet,
+		txhashset: &TxHashSet,
 	) -> Result<(), Error> {
 		debug!("validate_kernel_history: rewinding and validating kernel history (readonly)");
 
@@ -1151,11 +1152,11 @@ impl Chain {
 			self.validate_kernel_history(&header, &txhashset)?;
 
 			let header_pmmr = self.header_pmmr.read();
-			let batch = self.store.batch()?;
+			let mut batch = self.store.batch()?;
 			txhashset.verify_kernel_pos_index(
 				&self.genesis.header,
 				&header_pmmr,
-				&batch,
+				&mut batch,
 				None,
 				None,
 			)?;
@@ -1213,10 +1214,10 @@ impl Chain {
 		}
 
 		// Rebuild our output_pos index in the db based on fresh UTXO set.
-		txhashset.init_output_pos_index(&header_pmmr, &batch)?;
+		txhashset.init_output_pos_index(&header_pmmr, &mut batch)?;
 
 		// Rebuild our NRD kernel_pos index based on recent kernel history.
-		txhashset.init_recent_kernel_pos_index(&header_pmmr, &batch)?;
+		txhashset.init_recent_kernel_pos_index(&header_pmmr, &mut batch)?;
 
 		// Commit all the changes to the db.
 		batch.commit()?;
@@ -1257,9 +1258,9 @@ impl Chain {
 	/// *Only* runs if we are not in archive mode.
 	fn remove_historical_blocks(
 		&self,
-		header_pmmr: &txhashset::PMMRHandle<BlockHeader>,
+		header_pmmr: &PMMRHandle<BlockHeader>,
 		archive_header: BlockHeader,
-		batch: &store::Batch<'_>,
+		batch: &mut Batch<'_>,
 	) -> Result<(), Error> {
 		if self.archive_mode() {
 			return Ok(());
@@ -1293,17 +1294,24 @@ impl Chain {
 			return Ok(());
 		}
 
-		let mut count = 0;
 		let tail_hash = header_pmmr.get_header_hash_by_height(head.height - horizon)?;
 		let tail = batch.get_block_header(&tail_hash)?;
 
-		// 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;
+		// Remove old blocks (including short-lived fork blocks) which height < tail.height
+		let mut blocks_to_delete = vec![];
+		let iter = batch.blocks_iter()?;
+		for block in iter {
+			if let Ok(block) = block {
+				if block.header.height < tail.height {
+					blocks_to_delete.push(block.hash());
+				}
 			}
 		}
+		let mut count = 0;
+		for bh in blocks_to_delete {
+			let _ = batch.delete_block(&bh);
+			count += 1;
+		}
 
 		batch.save_body_tail(&Tip::from_header(&tail))?;
 
@@ -1345,7 +1353,7 @@ impl Chain {
 		// Take a write lock on the txhashet and start a new writeable db batch.
 		let header_pmmr = self.header_pmmr.read();
 		let mut txhashset = self.txhashset.write();
-		let batch = self.store.batch()?;
+		let mut batch = self.store.batch()?;
 
 		// Compact the txhashset itself (rewriting the pruned backend files).
 		{
@@ -1361,15 +1369,15 @@ impl Chain {
 
 		// If we are not in archival mode remove historical blocks from the db.
 		if !self.archive_mode() {
-			self.remove_historical_blocks(&header_pmmr, archive_header, &batch)?;
+			self.remove_historical_blocks(&header_pmmr, archive_header, &mut batch)?;
 		}
 
 		// Make sure our output_pos index is consistent with the UTXO set.
-		txhashset.init_output_pos_index(&header_pmmr, &batch)?;
+		txhashset.init_output_pos_index(&header_pmmr, &mut batch)?;
 
 		// TODO - Why is this part of chain compaction?
 		// Rebuild our NRD kernel_pos index based on recent kernel history.
-		txhashset.init_recent_kernel_pos_index(&header_pmmr, &batch)?;
+		txhashset.init_recent_kernel_pos_index(&header_pmmr, &mut batch)?;
 
 		// Commit all the above db changes.
 		batch.commit()?;