Save pending local transactions in the database (#4566)

* Create new column family for local node info

* remove DBTransaction::new reliance on DB

* KeyValueDB trait

* InMemory KeyValueDB implementation

* journaldb generic over KVDB

* make most of `ethcore` generic over KVDB

* fix json tests compilation

* get all tests compiling

* implement local store (just for transactions)

* finish local store API, test

* put everything into place

* better test for skipping bad transactions

* fix warning

* update local store every 15 minutes

* remove superfluous `{}`s

parity/main.rs

@@ -59,6 +59,7 @@ extern crate parity_hash_fetch as hash_fetch;
 extern crate parity_ipfs_api;
 extern crate parity_reactor;
 extern crate parity_updater as updater;
+extern crate parity_local_store as local_store;
 extern crate rpc_cli;
 
 #[macro_use]

parity/migration.rs

@@ -30,7 +30,7 @@ use ethcore::migrations::Extract;
 /// Database is assumed to be at default version, when no version file is found.
 const DEFAULT_VERSION: u32 = 5;
 /// Current version of database models.
-const CURRENT_VERSION: u32 = 10;
+const CURRENT_VERSION: u32 = 11;
 /// First version of the consolidated database.
 const CONSOLIDATION_VERSION: u32 = 9;
 /// Defines how many items are migrated to the new version of database at once.
@@ -146,6 +146,7 @@ pub fn default_migration_settings(compaction_profile: &CompactionProfile) -> Mig
 fn consolidated_database_migrations(compaction_profile: &CompactionProfile) -> Result<MigrationManager, Error> {
 	let mut manager = MigrationManager::new(default_migration_settings(compaction_profile));
 	manager.add_migration(migrations::ToV10::new()).map_err(|_| Error::MigrationImpossible)?;
+	manager.add_migration(migrations::ToV11::default()).map_err(|_| Error::MigrationImpossible)?;
 	Ok(manager)
 }
 

parity/run.rs

@@ -136,6 +136,21 @@ pub fn open_dapp(dapps_conf: &dapps::Configuration, dapp: &str) -> Result<(), St
 	Ok(())
 }
 
+// node info fetcher for the local store.
+struct FullNodeInfo {
+	miner: Arc<Miner>, // TODO: only TXQ needed, just use that after decoupling.
+}
+
+impl ::local_store::NodeInfo for FullNodeInfo {
+	fn pending_transactions(&self) -> Vec<::ethcore::transaction::PendingTransaction> {
+		let local_txs = self.miner.local_transactions();
+		self.miner.pending_transactions()
+			.into_iter()
+			.chain(self.miner.future_transactions())
+			.filter(|tx| local_txs.contains_key(&tx.hash()))
+			.collect()
+	}
+}
 
 pub fn execute(cmd: RunCmd, can_restart: bool, logger: Arc<RotatingLogger>) -> Result<bool, String> {
 	if cmd.ui && cmd.dapps_conf.enabled {
@@ -318,6 +333,33 @@ pub fn execute(cmd: RunCmd, can_restart: bool, logger: Arc<RotatingLogger>) -> R
 	let client = service.client();
 	let snapshot_service = service.snapshot_service();
 
+	// initialize the local node information store.
+	let store = {
+		let db = service.db();
+		let node_info = FullNodeInfo {
+			miner: miner.clone(),
+		};
+
+		let store = ::local_store::create(db, ::ethcore::db::COL_NODE_INFO, node_info);
+
+		// re-queue pending transactions.
+		match store.pending_transactions() {
+			Ok(pending) => {
+				for pending_tx in pending {
+					if let Err(e) = miner.import_own_transaction(&*client, pending_tx) {
+						warn!("Error importing saved transaction: {}", e)
+					}
+				}
+			}
+			Err(e) => warn!("Error loading cached pending transactions from disk: {}", e),
+		}
+
+		Arc::new(store)
+	};
+
+	// register it as an IO service to update periodically.
+	service.register_io_handler(store).map_err(|_| "Unable to register local store handler".to_owned())?;
+
 	// create external miner
 	let external_miner = Arc::new(ExternalMiner::default());