Missing blocks in filter_changes RPC

rpc/src/v1/helpers/poll_filter.rs

@@ -17,7 +17,7 @@
 //! Helper type with all filter state data.
 
 use std::{
-	collections::{BTreeSet, HashSet},
+	collections::{BTreeSet, HashSet, VecDeque},
 	sync::Arc,
 };
 use ethereum_types::H256;
@@ -49,7 +49,11 @@ impl SyncPollFilter {
 #[derive(Clone)]
 pub enum PollFilter {
 	/// Number of last block which client was notified about.
-	Block(BlockNumber),
+	Block {
+		last_block_number: BlockNumber,
+		#[doc(hidden)]
+		recent_reported_hashes: VecDeque<(BlockNumber, H256)>,
+	},
 	/// Hashes of all pending transactions the client knows about.
 	PendingTransaction(BTreeSet<H256>),
 	/// Number of From block number, last seen block hash, pending logs and log filter itself.
@@ -62,6 +66,10 @@ pub enum PollFilter {
 	}
 }
 
+impl PollFilter {
+	pub (in v1) const MAX_BLOCK_HISTORY_SIZE: usize = 32;
+}
+
 /// Returns only last `n` logs
 pub fn limit_logs(mut logs: Vec<Log>, limit: Option<usize>) -> Vec<Log> {
 	let len = logs.len();

rpc/src/v1/impls/eth_filter.rs

@@ -17,7 +17,7 @@
 //! Eth Filter RPC implementation
 
 use std::sync::Arc;
-use std::collections::BTreeSet;
+use std::collections::{BTreeSet, VecDeque};
 
 use ethcore::miner::{self, MinerService};
 use ethcore::filter::Filter as EthcoreFilter;
@@ -153,7 +153,10 @@ impl<T: Filterable + Send + Sync + 'static> EthFilter for T {
 	fn new_block_filter(&self) -> Result<RpcU256> {
 		let mut polls = self.polls().lock();
 		// +1, since we don't want to include the current block
-		let id = polls.create_poll(SyncPollFilter::new(PollFilter::Block(self.best_block_number() + 1)));
+		let id = polls.create_poll(SyncPollFilter::new(PollFilter::Block {
+			last_block_number: self.best_block_number(),
+			recent_reported_hashes: VecDeque::with_capacity(PollFilter::MAX_BLOCK_HISTORY_SIZE),
+		}));
 		Ok(id.into())
 	}
 
@@ -171,15 +174,33 @@ impl<T: Filterable + Send + Sync + 'static> EthFilter for T {
 		};
 
 		Box::new(filter.modify(|filter| match *filter {
-			PollFilter::Block(ref mut block_number) => {
-				// +1, cause we want to return hashes including current block hash.
-				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).map(Into::into))
-					.collect::<Vec<RpcH256>>();
-
-				*block_number = current_number;
+			PollFilter::Block {
+				ref mut last_block_number,
+				ref mut recent_reported_hashes,
+			} => {
+				// Check validity of recently reported blocks -- in case of re-org, rewind block to last valid
+				while let Some((num, hash)) = recent_reported_hashes.front().cloned() {
+					if self.block_hash(BlockId::Number(num)) == Some(hash) { break; }
+					*last_block_number = num - 1;
+					recent_reported_hashes.pop_front();
+				}
+				let current_number = self.best_block_number();
+				let mut hashes = Vec::new();
+				for n in (*last_block_number + 1)..=current_number {
+					let block_number = BlockId::Number(n);
+					match self.block_hash(block_number) {
+						Some(hash) => {
+							*last_block_number = n;
+							hashes.push(RpcH256::from(hash));
+							// Only keep the most recent history
+							if recent_reported_hashes.len() >= PollFilter::MAX_BLOCK_HISTORY_SIZE {
+								recent_reported_hashes.pop_back();
+							}
+							recent_reported_hashes.push_front((n, hash));
+						},
+						None => (),
+					}
+				}
 
 				Either::A(future::ok(FilterChanges::Hashes(hashes)))
 			},

rpc/src/v1/tests/mocked/eth.rs

@@ -317,10 +317,26 @@ fn rpc_blocks_filter() {
 
 	tester.client.add_blocks(2, EachBlockWith::Nothing);
 
+	let hash1 = tester.client.block_hash(BlockId::Number(1)).unwrap();
+	let hash2 = tester.client.block_hash(BlockId::Number(2)).unwrap();
 	let response = format!(
 		r#"{{"jsonrpc":"2.0","result":["0x{:x}","0x{:x}"],"id":1}}"#,
-		tester.client.block_hash(BlockId::Number(1)).unwrap(),
-		tester.client.block_hash(BlockId::Number(2)).unwrap());
+		hash1,
+		hash2);
+
+	assert_eq!(tester.io.handle_request_sync(request_changes), Some(response.to_owned()));
+
+	// in the case of a re-org we get same block number if hash is different - BlockId::Number(2)
+	tester.client.blocks.write().remove(&hash2).unwrap();
+	tester.client.numbers.write().remove(&2).unwrap();
+	*tester.client.last_hash.write() = hash1;
+	tester.client.add_blocks(2, EachBlockWith::Uncle);
+
+	let request_changes = r#"{"jsonrpc": "2.0", "method": "eth_getFilterChanges", "params": ["0x0"], "id": 2}"#;
+	let response = format!(
+		r#"{{"jsonrpc":"2.0","result":["0x{:x}","0x{:x}"],"id":2}}"#,
+		tester.client.block_hash(BlockId::Number(2)).unwrap(),
+		tester.client.block_hash(BlockId::Number(3)).unwrap());
 
 	assert_eq!(tester.io.handle_request_sync(request_changes), Some(response.to_owned()));
 }