Fix parallel transactions race-condition

ethcore/engines/instant-seal/src/lib.rs

@@ -14,6 +14,8 @@
 // You should have received a copy of the GNU General Public License
 // along with Parity Ethereum.  If not, see <http://www.gnu.org/licenses/>.
 
+use std::sync::atomic::{AtomicU64, Ordering};
+
 use common_types::{
 	header::Header,
 	engines::{
@@ -51,6 +53,7 @@ impl From<ethjson::spec::InstantSealParams> for InstantSealParams {
 pub struct InstantSeal {
 	params: InstantSealParams,
 	machine: Machine,
+	last_sealed_block: AtomicU64,
 }
 
 impl InstantSeal {
@@ -59,6 +62,7 @@ impl InstantSeal {
 		InstantSeal {
 			params,
 			machine,
+			last_sealed_block: AtomicU64::new(0),
 		}
 	}
 }
@@ -71,11 +75,19 @@ impl Engine for InstantSeal {
 	fn sealing_state(&self) -> SealingState { SealingState::Ready }
 
 	fn generate_seal(&self, block: &ExecutedBlock, _parent: &Header) -> Seal {
-		if block.transactions.is_empty() {
-			Seal::None
-		} else {
-			Seal::Regular(Vec::new())
+		if !block.transactions.is_empty() {
+			let block_number = block.header.number();
+			let last_sealed_block = self.last_sealed_block.load(Ordering::SeqCst);
+			// Return a regular seal if the given block is _higher_ than
+			// the last sealed one
+			if block_number > last_sealed_block {
+				let prev_last_sealed_block = self.last_sealed_block.compare_and_swap(last_sealed_block, block_number, Ordering::SeqCst);
+				if prev_last_sealed_block == last_sealed_block {
+					return Seal::Regular(Vec::new())
+				}
+			}
 		}
+		Seal::None
 	}
 
 	fn verify_local_seal(&self, _header: &Header) -> Result<(), Error> {

ethcore/src/client/client.rs

@@ -1268,6 +1268,11 @@ impl Client {
 			_   => self.block_header(id).and_then(|h| h.decode().ok())
 		}
 	}
+
+	/// Maybe call `update_sealing` method on the Miner
+	pub fn maybe_update_sealing(&self) {
+		self.importer.miner.maybe_update_sealing(self);
+	}
 }
 
 impl DatabaseRestore for Client {

ethcore/src/miner/miner.rs

@@ -33,7 +33,7 @@ use futures::sync::mpsc;
 use io::IoChannel;
 use miner::filter_options::{FilterOptions, FilterOperator};
 use miner::pool_client::{PoolClient, CachedNonceClient, NonceCache};
-use miner;
+use miner::{self, MinerService};
 use parking_lot::{Mutex, RwLock};
 use rayon::prelude::*;
 use types::{
@@ -859,9 +859,9 @@ impl Miner {
 			false
 		}
 	}
+
 	/// Prepare pending block, check whether sealing is needed, and then update sealing.
 	fn prepare_and_update_sealing<C: miner::BlockChainClient>(&self, chain: &C) {
-		use miner::MinerService;
 		match self.engine.sealing_state() {
 			SealingState::Ready => {
 				self.maybe_enable_sealing();
@@ -876,6 +876,15 @@ impl Miner {
 			SealingState::NotReady => { self.maybe_enable_sealing(); },
 		}
 	}
+
+	/// Call `update_sealing` if there are some local pending transactions.
+	/// This should be called after an internal block import, which might not have
+	/// included all the pending local transactions.
+	pub fn maybe_update_sealing<C: miner::BlockChainClient>(&self, chain: &C) {
+		if self.transaction_queue.has_local_pending_transactions() {
+			self.update_sealing(chain);
+		}
+	}
 }
 
 impl miner::MinerService for Miner {
@@ -1429,15 +1438,22 @@ impl miner::MinerService for Miner {
 						service_transaction_checker.as_ref(),
 					);
 					queue.cull(client);
+					if is_internal_import {
+						chain.maybe_update_sealing();
+					}
 				};
 
 				if let Err(e) = channel.send(ClientIoMessage::<Client>::execute(cull)) {
 					warn!(target: "miner", "Error queueing cull: {:?}", e);
 				}
 			} else {
 				self.transaction_queue.cull(client);
+				if is_internal_import {
+					self.maybe_update_sealing(chain);
+				}
 			}
 		}
+
 		if let Some(ref service_transaction_checker) = self.service_transaction_checker {
 			match service_transaction_checker.refresh_cache(chain) {
 				Ok(true) => {