Handle removed logs in filter changes and add geth compatibility field (#8796)

* Add removed geth compatibility field in log

* Fix mocked tests

* Add field block hash in PollFilter

* Store last block hash info for log filters

* Implement canon route

* Use canon logs for fetching reorg logs

Light client removed logs fetching is disabled. It looks expensive.

* Make sure removed flag is set

* Address grumbles

rpc/src/v1/impls/eth_filter.rs

@@ -39,7 +39,7 @@ pub trait Filterable {
 	fn best_block_number(&self) -> u64;
 
 	/// Get a block hash by block id.
-	fn block_hash(&self, id: BlockId) -> Option<RpcH256>;
+	fn block_hash(&self, id: BlockId) -> Option<H256>;
 
 	/// pending transaction hashes at the given block.
 	fn pending_transactions_hashes(&self) -> Vec<H256>;
@@ -52,6 +52,9 @@ pub trait Filterable {
 
 	/// Get a reference to the poll manager.
 	fn polls(&self) -> &Mutex<PollManager<PollFilter>>;
+
+	/// Get removed logs within route from the given block to the nearest canon block, not including the canon block. Also returns how many logs have been traversed.
+	fn removed_logs(&self, block_hash: H256, filter: &EthcoreFilter) -> (Vec<Log>, u64);
 }
 
 /// Eth filter rpc implementation for a full node.
@@ -80,8 +83,8 @@ impl<C, M> Filterable for EthFilterClient<C, M> where
 		self.client.chain_info().best_block_number
 	}
 
-	fn block_hash(&self, id: BlockId) -> Option<RpcH256> {
-		self.client.block_hash(id).map(Into::into)
+	fn block_hash(&self, id: BlockId) -> Option<H256> {
+		self.client.block_hash(id)
 	}
 
 	fn pending_transactions_hashes(&self) -> Vec<H256> {
@@ -100,13 +103,47 @@ impl<C, M> Filterable for EthFilterClient<C, M> where
 	}
 
 	fn polls(&self) -> &Mutex<PollManager<PollFilter>> { &self.polls }
+
+	fn removed_logs(&self, block_hash: H256, filter: &EthcoreFilter) -> (Vec<Log>, u64) {
+		let inner = || -> Option<Vec<H256>> {
+			let mut route = Vec::new();
+
+			let mut current_block_hash = block_hash;
+			let mut current_block_header = self.client.block_header(BlockId::Hash(current_block_hash))?;
+
+			while current_block_hash != self.client.block_hash(BlockId::Number(current_block_header.number()))? {
+				route.push(current_block_hash);
+
+				current_block_hash = current_block_header.parent_hash();
+				current_block_header = self.client.block_header(BlockId::Hash(current_block_hash))?;
+			}
+
+			Some(route)
+		};
+
+		let route = inner().unwrap_or_default();
+		let route_len = route.len() as u64;
+		(route.into_iter().flat_map(|block_hash| {
+			let mut filter = filter.clone();
+			filter.from_block = BlockId::Hash(block_hash);
+			filter.to_block = filter.from_block;
+
+			self.client.logs(filter).into_iter().map(|log| {
+				let mut log: Log = log.into();
+				log.log_type = "removed".into();
+				log.removed = true;
+
+				log
+			})
+		}).collect(), route_len)
+	}
 }
 
 impl<T: Filterable + Send + Sync + 'static> EthFilter for T {
 	fn new_filter(&self, filter: Filter) -> Result<RpcU256> {
 		let mut polls = self.polls().lock();
 		let block_number = self.best_block_number();
-		let id = polls.create_poll(PollFilter::Logs(block_number, Default::default(), filter));
+		let id = polls.create_poll(PollFilter::Logs(block_number, None, Default::default(), filter));
 		Ok(id.into())
 	}
 
@@ -134,7 +171,7 @@ impl<T: Filterable + Send + Sync + 'static> EthFilter for T {
 					let current_number = self.best_block_number() + 1;
 					let hashes = (*block_number..current_number).into_iter()
 						.map(BlockId::Number)
-						.filter_map(|id| self.block_hash(id))
+						.filter_map(|id| self.block_hash(id).map(Into::into))
 						.collect::<Vec<RpcH256>>();
 
 					*block_number = current_number;
@@ -164,7 +201,7 @@ impl<T: Filterable + Send + Sync + 'static> EthFilter for T {
 					// return new hashes
 					Either::A(future::ok(FilterChanges::Hashes(new_hashes)))
 				},
-				PollFilter::Logs(ref mut block_number, ref mut previous_logs, ref filter) => {
+				PollFilter::Logs(ref mut block_number, ref mut last_block_hash, ref mut previous_logs, ref filter) => {
 					// retrive the current block number
 					let current_number = self.best_block_number();
 
@@ -173,6 +210,11 @@ impl<T: Filterable + Send + Sync + 'static> EthFilter for T {
 
 					// build appropriate filter
 					let mut filter: EthcoreFilter = filter.clone().into();
+
+					// retrieve reorg logs
+					let (mut reorg, reorg_len) = last_block_hash.map_or_else(|| (Vec::new(), 0), |h| self.removed_logs(h, &filter));
+					*block_number -= reorg_len as u64;
+
 					filter.from_block = BlockId::Number(*block_number);
 					filter.to_block = BlockId::Latest;
 
@@ -198,9 +240,14 @@ impl<T: Filterable + Send + Sync + 'static> EthFilter for T {
 					// we want to get logs
 					*block_number = current_number + 1;
 
+					// save the current block hash, which we used to get back to the
+					// canon chain in case of reorg.
+					*last_block_hash = self.block_hash(BlockId::Number(current_number));
+
 					// retrieve logs in range from_block..min(BlockId::Latest..to_block)
 					let limit = filter.limit;
 					Either::B(self.logs(filter)
+						.map(move |logs| { reorg.extend(logs); reorg }) // append reorg logs in the front
 						.map(move |mut logs| { logs.extend(pending); logs }) // append fetched pending logs
 						.map(move |logs| limit_logs(logs, limit)) // limit the logs
 						.map(FilterChanges::Logs))
@@ -214,7 +261,7 @@ impl<T: Filterable + Send + Sync + 'static> EthFilter for T {
 			let mut polls = self.polls().lock();
 
 			match polls.poll(&index.value()) {
-				Some(&PollFilter::Logs(ref _block_number, ref _previous_log, ref filter)) => filter.clone(),
+				Some(&PollFilter::Logs(ref _block_number, ref _last_block_hash, ref _previous_log, ref filter)) => filter.clone(),
 				// just empty array
 				Some(_) => return Box::new(future::ok(Vec::new())),
 				None => return Box::new(future::err(errors::filter_not_found())),

rpc/src/v1/impls/eth_pubsub.rs

@@ -256,6 +256,7 @@ impl<C: BlockChainClient> ChainNotify for ChainNotificationHandler<C> {
 				&ChainRouteType::Retracted =>
 					Ok(self.client.logs(filter).into_iter().map(Into::into).map(|mut log: Log| {
 						log.log_type = "removed".into();
+						log.removed = true;
 						log
 					}).collect()),
 			}

rpc/src/v1/impls/light/eth.rs

@@ -529,8 +529,8 @@ impl<T: LightChainClient + 'static> Eth for EthClient<T> {
 impl<T: LightChainClient + 'static> Filterable for EthClient<T> {
 	fn best_block_number(&self) -> u64 { self.client.chain_info().best_block_number }
 
-	fn block_hash(&self, id: BlockId) -> Option<RpcH256> {
-		self.client.block_hash(id).map(Into::into)
+	fn block_hash(&self, id: BlockId) -> Option<::ethereum_types::H256> {
+		self.client.block_hash(id)
 	}
 
 	fn pending_transactions_hashes(&self) -> Vec<::ethereum_types::H256> {
@@ -548,6 +548,10 @@ impl<T: LightChainClient + 'static> Filterable for EthClient<T> {
 	fn polls(&self) -> &Mutex<PollManager<PollFilter>> {
 		&self.polls
 	}
+
+	fn removed_logs(&self, _block_hash: ::ethereum_types::H256, _filter: &EthcoreFilter) -> (Vec<Log>, u64) {
+		(Default::default(), 0)
+	}
 }
 
 fn extract_uncle_at_index<T: LightChainClient>(block: encoded::Block, index: Index, client: Arc<T>) -> Option<RichBlock> {