Better support for eth_getLogs in light mode (#9186)

* Light client on-demand request for headers range.

* Cache headers in HeaderWithAncestors response.

Also fulfills request locally if all headers are in cache.

* LightFetch::logs fetches missing headers on demand.

* LightFetch::logs limit the number of headers requested at a time.

* LightFetch::logs refactor header fetching logic.

* Enforce limit on header range length in light client logs request.

* Fix light request tests after struct change.

* Respond to review comments.

ethcore/light/src/lib.rs

@@ -43,7 +43,7 @@ pub mod provider;
 mod types;
 
 pub use self::cache::Cache;
-pub use self::provider::Provider;
+pub use self::provider::{Provider, MAX_HEADERS_PER_REQUEST};
 pub use self::transaction_queue::TransactionQueue;
 pub use types::request as request;
 

ethcore/light/src/on_demand/mod.rs

@@ -204,6 +204,8 @@ fn guess_capabilities(requests: &[CheckedRequest]) -> Capabilities {
 				caps.serve_headers = true,
 			CheckedRequest::HeaderByHash(_, _) =>
 				caps.serve_headers = true,
+			CheckedRequest::HeaderWithAncestors(_, _) =>
+				caps.serve_headers = true,
 			CheckedRequest::TransactionIndex(_, _) => {} // hashes yield no info.
 			CheckedRequest::Signal(_, _) =>
 				caps.serve_headers = true,

ethcore/light/src/on_demand/request.rs

@@ -16,6 +16,7 @@
 
 //! Request types, verification, and verification errors.
 
+use std::cmp;
 use std::sync::Arc;
 
 use bytes::Bytes;
@@ -47,6 +48,8 @@ pub enum Request {
 	HeaderProof(HeaderProof),
 	/// A request for a header by hash.
 	HeaderByHash(HeaderByHash),
+	/// A request for a header by hash with a range of its ancestors.
+	HeaderWithAncestors(HeaderWithAncestors),
 	/// A request for the index of a transaction.
 	TransactionIndex(TransactionIndex),
 	/// A request for block receipts.
@@ -136,6 +139,7 @@ macro_rules! impl_single {
 // implement traits for each kind of request.
 impl_single!(HeaderProof, HeaderProof, (H256, U256));
 impl_single!(HeaderByHash, HeaderByHash, encoded::Header);
+impl_single!(HeaderWithAncestors, HeaderWithAncestors, Vec<encoded::Header>);
 impl_single!(TransactionIndex, TransactionIndex, net_request::TransactionIndexResponse);
 impl_single!(Receipts, BlockReceipts, Vec<Receipt>);
 impl_single!(Body, Body, encoded::Block);
@@ -246,6 +250,7 @@ impl From<encoded::Header> for HeaderRef {
 pub enum CheckedRequest {
 	HeaderProof(HeaderProof, net_request::IncompleteHeaderProofRequest),
 	HeaderByHash(HeaderByHash, net_request::IncompleteHeadersRequest),
+	HeaderWithAncestors(HeaderWithAncestors, net_request::IncompleteHeadersRequest),
 	TransactionIndex(TransactionIndex, net_request::IncompleteTransactionIndexRequest),
 	Receipts(BlockReceipts, net_request::IncompleteReceiptsRequest),
 	Body(Body, net_request::IncompleteBodyRequest),
@@ -267,6 +272,16 @@ impl From<Request> for CheckedRequest {
 				};
 				CheckedRequest::HeaderByHash(req, net_req)
 			}
+			Request::HeaderWithAncestors(req) => {
+				let net_req = net_request::IncompleteHeadersRequest {
+					start: req.block_hash.map(Into::into),
+					skip: 0,
+					max: req.ancestor_count + 1,
+					reverse: true,
+				};
+				trace!(target: "on_demand", "HeaderWithAncestors Request, {:?}", net_req);
+				CheckedRequest::HeaderWithAncestors(req, net_req)
+			}
 			Request::HeaderProof(req) => {
 				let net_req = net_request::IncompleteHeaderProofRequest {
 					num: req.num().into(),
@@ -335,6 +350,7 @@ impl CheckedRequest {
 		match self {
 			CheckedRequest::HeaderProof(_, req) => NetRequest::HeaderProof(req),
 			CheckedRequest::HeaderByHash(_, req) => NetRequest::Headers(req),
+			CheckedRequest::HeaderWithAncestors(_, req) => NetRequest::Headers(req),
 			CheckedRequest::TransactionIndex(_, req) => NetRequest::TransactionIndex(req),
 			CheckedRequest::Receipts(_, req) => NetRequest::Receipts(req),
 			CheckedRequest::Body(_, req) => NetRequest::Body(req),
@@ -390,6 +406,27 @@ impl CheckedRequest {
 
 				None
 			}
+			CheckedRequest::HeaderWithAncestors(_, ref req) => {
+				if req.skip != 1 || !req.reverse {
+					return None;
+				}
+
+				if let Some(&net_request::HashOrNumber::Hash(start)) = req.start.as_ref() {
+					let mut result = Vec::with_capacity(req.max as usize);
+					let mut hash = start;
+					let mut cache = cache.lock();
+					for _ in 0..req.max {
+						match cache.block_header(&hash) {
+							Some(header) => {
+								hash = header.parent_hash();
+								result.push(header);
+							}
+							None => return None,
+						}
+					}
+					Some(Response::HeaderWithAncestors(result))
+				} else { None }
+			}
 			CheckedRequest::Receipts(ref check, ref req) => {
 				// empty transactions -> no receipts
 				if check.0.as_ref().ok().map_or(false, |hdr| hdr.receipts_root() == KECCAK_NULL_RLP) {
@@ -458,6 +495,7 @@ macro_rules! match_me {
 		match $me {
 			CheckedRequest::HeaderProof($check, $req) => $e,
 			CheckedRequest::HeaderByHash($check, $req) => $e,
+			CheckedRequest::HeaderWithAncestors($check, $req) => $e,
 			CheckedRequest::TransactionIndex($check, $req) => $e,
 			CheckedRequest::Receipts($check, $req) => $e,
 			CheckedRequest::Body($check, $req) => $e,
@@ -487,6 +525,15 @@ impl IncompleteRequest for CheckedRequest {
 					_ => Ok(()),
 				}
 			}
+			CheckedRequest::HeaderWithAncestors(ref check, ref req) => {
+				req.check_outputs(&mut f)?;
+
+				// make sure the output given is definitively a hash.
+				match check.block_hash {
+					Field::BackReference(r, idx) => f(r, idx, OutputKind::Hash),
+					_ => Ok(()),
+				}
+			}
 			CheckedRequest::TransactionIndex(_, ref req) => req.check_outputs(f),
 			CheckedRequest::Receipts(_, ref req) => req.check_outputs(f),
 			CheckedRequest::Body(_, ref req) => req.check_outputs(f),
@@ -507,15 +554,46 @@ impl IncompleteRequest for CheckedRequest {
 
 	fn complete(self) -> Result<Self::Complete, net_request::NoSuchOutput> {
 		match self {
-			CheckedRequest::HeaderProof(_, req) => req.complete().map(CompleteRequest::HeaderProof),
-			CheckedRequest::HeaderByHash(_, req) => req.complete().map(CompleteRequest::Headers),
-			CheckedRequest::TransactionIndex(_, req) => req.complete().map(CompleteRequest::TransactionIndex),
-			CheckedRequest::Receipts(_, req) => req.complete().map(CompleteRequest::Receipts),
-			CheckedRequest::Body(_, req) => req.complete().map(CompleteRequest::Body),
-			CheckedRequest::Account(_, req) => req.complete().map(CompleteRequest::Account),
-			CheckedRequest::Code(_, req) => req.complete().map(CompleteRequest::Code),
-			CheckedRequest::Execution(_, req) => req.complete().map(CompleteRequest::Execution),
-			CheckedRequest::Signal(_, req) => req.complete().map(CompleteRequest::Signal),
+			CheckedRequest::HeaderProof(_, req) => {
+				trace!(target: "on_demand", "HeaderProof request completed {:?}", req);
+				req.complete().map(CompleteRequest::HeaderProof)
+			}
+			CheckedRequest::HeaderByHash(_, req) => {
+				trace!(target: "on_demand", "HeaderByHash request completed {:?}", req);
+				req.complete().map(CompleteRequest::Headers)
+			}
+			CheckedRequest::HeaderWithAncestors(_, req) => {
+				trace!(target: "on_demand", "HeaderWithAncestors request completed {:?}", req);
+				req.complete().map(CompleteRequest::Headers)
+			}
+			CheckedRequest::TransactionIndex(_, req) => {
+				trace!(target: "on_demand", "TransactionIndex request completed {:?}", req);
+				req.complete().map(CompleteRequest::TransactionIndex)
+			}
+			CheckedRequest::Receipts(_, req) => {
+				trace!(target: "on_demand", "Receipt request completed {:?}", req);
+				req.complete().map(CompleteRequest::Receipts)
+			}
+			CheckedRequest::Body(_, req) => {
+				trace!(target: "on_demand", "Block request completed {:?}", req);
+				req.complete().map(CompleteRequest::Body)
+			}
+			CheckedRequest::Account(_, req) => {
+				trace!(target: "on_demand", "Account request completed {:?}", req);
+				req.complete().map(CompleteRequest::Account)
+			}
+			CheckedRequest::Code(_, req) => {
+				trace!(target: "on_demand", "Code request completed {:?}", req);
+				req.complete().map(CompleteRequest::Code)
+			}
+			CheckedRequest::Execution(_, req) => {
+				trace!(target: "on_demand", "Execution request completed {:?}", req);
+				req.complete().map(CompleteRequest::Execution)
+			}
+			CheckedRequest::Signal(_, req) => {
+				trace!(target: "on_demand", "Signal request completed {:?}", req);
+				req.complete().map(CompleteRequest::Signal)
+			}
 		}
 	}
 
@@ -551,6 +629,9 @@ impl net_request::CheckedRequest for CheckedRequest {
 			CheckedRequest::HeaderByHash(ref prover, _) =>
 				expect!((&NetResponse::Headers(ref res), &CompleteRequest::Headers(ref req)) =>
 					prover.check_response(cache, &req.start, &res.headers).map(Response::HeaderByHash)),
+			CheckedRequest::HeaderWithAncestors(ref prover, _) =>
+				expect!((&NetResponse::Headers(ref res), &CompleteRequest::Headers(ref req)) =>
+					prover.check_response(cache, &req.start, &res.headers).map(Response::HeaderWithAncestors)),
 			CheckedRequest::TransactionIndex(ref prover, _) =>
 				expect!((&NetResponse::TransactionIndex(ref res), _) =>
 					prover.check_response(cache, res).map(Response::TransactionIndex)),
@@ -584,6 +665,8 @@ pub enum Response {
 	HeaderProof((H256, U256)),
 	/// Response to a header-by-hash request.
 	HeaderByHash(encoded::Header),
+	/// Response to a header-by-hash with ancestors request.
+	HeaderWithAncestors(Vec<encoded::Header>),
 	/// Response to a transaction-index request.
 	TransactionIndex(net_request::TransactionIndexResponse),
 	/// Response to a receipts request.
@@ -625,6 +708,10 @@ pub enum Error {
 	Decoder(::rlp::DecoderError),
 	/// Empty response.
 	Empty,
+	/// Response data length exceeds request max.
+	TooManyResults(u64, u64),
+	/// Response data is incomplete.
+	TooFewResults(u64, u64),
 	/// Trie lookup error (result of bad proof)
 	Trie(TrieError),
 	/// Bad inclusion proof
@@ -641,6 +728,8 @@ pub enum Error {
 	WrongTrieRoot(H256, H256),
 	/// Wrong response kind.
 	WrongKind,
+	/// Wrong sequence of headers.
+	WrongHeaderSequence,
 }
 
 impl From<::rlp::DecoderError> for Error {
@@ -701,6 +790,65 @@ impl HeaderProof {
 	}
 }
 
+/// Request for a header by hash with a range of ancestors.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct HeaderWithAncestors {
+	/// Hash of the last block in the range to fetch.
+	pub block_hash: Field<H256>,
+	/// Number of headers before the last block to fetch in addition.
+	pub ancestor_count: u64,
+}
+
+impl HeaderWithAncestors {
+	/// Check a response for the headers.
+	pub fn check_response(
+		&self,
+		cache: &Mutex<::cache::Cache>,
+		start: &net_request::HashOrNumber,
+		headers: &[encoded::Header]
+	) -> Result<Vec<encoded::Header>, Error> {
+		let expected_hash = match (self.block_hash, start) {
+			(Field::Scalar(ref h), &net_request::HashOrNumber::Hash(ref h2)) => {
+				if h != h2 { return Err(Error::WrongHash(*h, *h2)) }
+				*h
+			}
+			(_, &net_request::HashOrNumber::Hash(h2)) => h2,
+			_ => return Err(Error::HeaderByNumber),
+		};
+
+		let start_header = headers.first().ok_or(Error::Empty)?;
+		let start_hash = start_header.hash();
+		if start_hash != expected_hash {
+			return Err(Error::WrongHash(expected_hash, start_hash));
+		}
+
+		let expected_len = 1 + cmp::min(self.ancestor_count, start_header.number());
+		let actual_len = headers.len() as u64;
+		match actual_len.cmp(&expected_len) {
+			cmp::Ordering::Less =>
+				return Err(Error::TooFewResults(expected_len, actual_len)),
+			cmp::Ordering::Greater =>
+				return Err(Error::TooManyResults(expected_len, actual_len)),
+			cmp::Ordering::Equal => (),
+		};
+
+		for (header, prev_header) in headers.iter().zip(headers[1..].iter()) {
+			if header.number() != prev_header.number() + 1 ||
+				header.parent_hash() != prev_header.hash()
+			{
+				return Err(Error::WrongHeaderSequence)
+			}
+		}
+
+		let mut cache = cache.lock();
+		for header in headers {
+			cache.insert_block_header(header.hash(), header.clone());
+		}
+
+		Ok(headers.to_vec())
+	}
+}
+
 /// Request for a header by hash.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct HeaderByHash(pub Field<H256>);
@@ -993,6 +1141,83 @@ mod tests {
 		assert!(HeaderByHash(hash.into()).check_response(&cache, &hash.into(), &[raw_header]).is_ok())
 	}
 
+	#[test]
+	fn check_header_with_ancestors() {
+		let mut last_header_hash = H256::default();
+		let mut headers = (0..11).map(|num| {
+			let mut header = Header::new();
+			header.set_number(num);
+			header.set_parent_hash(last_header_hash);
+
+			last_header_hash = header.hash();
+			header
+		}).collect::<Vec<_>>();
+
+		headers.reverse();  // because responses are in reverse order
+
+		let raw_headers = headers.iter()
+			.map(|hdr| encoded::Header::new(::rlp::encode(hdr).into_vec()))
+			.collect::<Vec<_>>();
+
+		let mut invalid_successor = Header::new();
+		invalid_successor.set_number(11);
+		invalid_successor.set_parent_hash(headers[1].hash());
+
+		let raw_invalid_successor = encoded::Header::new(::rlp::encode(&invalid_successor).into_vec());
+
+		let cache = Mutex::new(make_cache());
+
+		let header_with_ancestors = |hash, count| {
+			HeaderWithAncestors {
+				block_hash: hash,
+				ancestor_count: count
+			}
+		};
+
+		// Correct responses
+		assert!(header_with_ancestors(headers[0].hash().into(), 0)
+				.check_response(&cache, &headers[0].hash().into(), &raw_headers[0..1]).is_ok());
+		assert!(header_with_ancestors(headers[0].hash().into(), 2)
+				.check_response(&cache, &headers[0].hash().into(), &raw_headers[0..3]).is_ok());
+		assert!(header_with_ancestors(headers[0].hash().into(), 10)
+				.check_response(&cache, &headers[0].hash().into(), &raw_headers[0..11]).is_ok());
+		assert!(header_with_ancestors(headers[2].hash().into(), 2)
+				.check_response(&cache, &headers[2].hash().into(), &raw_headers[2..5]).is_ok());
+		assert!(header_with_ancestors(headers[2].hash().into(), 10)
+				.check_response(&cache, &headers[2].hash().into(), &raw_headers[2..11]).is_ok());
+		assert!(header_with_ancestors(invalid_successor.hash().into(), 0)
+				.check_response(&cache, &invalid_successor.hash().into(), &[raw_invalid_successor.clone()]).is_ok());
+
+		// Incorrect responses
+		assert_eq!(header_with_ancestors(invalid_successor.hash().into(), 0)
+				   .check_response(&cache, &headers[0].hash().into(), &raw_headers[0..1]),
+				   Err(Error::WrongHash(invalid_successor.hash(), headers[0].hash())));
+		assert_eq!(header_with_ancestors(headers[0].hash().into(), 0)
+				   .check_response(&cache, &headers[0].hash().into(), &[]),
+				   Err(Error::Empty));
+		assert_eq!(header_with_ancestors(headers[0].hash().into(), 10)
+				   .check_response(&cache, &headers[0].hash().into(), &raw_headers[0..10]),
+				   Err(Error::TooFewResults(11, 10)));
+		assert_eq!(header_with_ancestors(headers[0].hash().into(), 9)
+				   .check_response(&cache, &headers[0].hash().into(), &raw_headers[0..11]),
+				   Err(Error::TooManyResults(10, 11)));
+
+		let response = &[raw_headers[0].clone(), raw_headers[2].clone()];
+		assert_eq!(header_with_ancestors(headers[0].hash().into(), 1)
+				   .check_response(&cache, &headers[0].hash().into(), response),
+				   Err(Error::WrongHeaderSequence));
+
+		let response = &[raw_invalid_successor.clone(), raw_headers[0].clone()];
+		assert_eq!(header_with_ancestors(invalid_successor.hash().into(), 1)
+				   .check_response(&cache, &invalid_successor.hash().into(), response),
+				   Err(Error::WrongHeaderSequence));
+
+		let response = &[raw_invalid_successor.clone(), raw_headers[1].clone()];
+		assert_eq!(header_with_ancestors(invalid_successor.hash().into(), 1)
+				   .check_response(&cache, &invalid_successor.hash().into(), response),
+				   Err(Error::WrongHeaderSequence));
+	}
+
 	#[test]
 	fn check_body() {
 		use rlp::RlpStream;