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),
@@ -268,6 +273,16 @@ impl From<Request> for CheckedRequest {
 				trace!(target: "on_demand", "HeaderByHash Request, {:?}", net_req);
 				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(),
@@ -344,6 +359,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),
@@ -399,6 +415,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) {
@@ -467,6 +504,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,
@@ -496,6 +534,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),
@@ -524,6 +571,10 @@ impl IncompleteRequest for CheckedRequest {
 				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)
@@ -587,6 +638,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)),
@@ -620,6 +674,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.
@@ -661,6 +717,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
@@ -677,6 +737,8 @@ pub enum Error {
 	WrongTrieRoot(H256, H256),
 	/// Wrong response kind.
 	WrongKind,
+	/// Wrong sequence of headers.
+	WrongHeaderSequence,
 }
 
 impl From<::rlp::DecoderError> for Error {
@@ -737,6 +799,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>);
@@ -1045,6 +1166,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;

ethcore/light/src/provider.rs

@@ -33,6 +33,9 @@ use transaction_queue::TransactionQueue;
 
 use request;
 
+/// Maximum allowed size of a headers request.
+pub const MAX_HEADERS_PER_REQUEST: u64 = 512;
+
 /// Defines the operations that a provider for the light subprotocol must fulfill.
 pub trait Provider: Send + Sync {
 	/// Provide current blockchain info.
@@ -54,7 +57,6 @@ pub trait Provider: Send + Sync {
 	/// results within must adhere to the `skip` and `reverse` parameters.
 	fn block_headers(&self, req: request::CompleteHeadersRequest) -> Option<request::HeadersResponse> {
 		use request::HashOrNumber;
-		const MAX_HEADERS_TO_SEND: u64 = 512;
 
 		if req.max == 0 { return None }
 
@@ -83,7 +85,7 @@ pub trait Provider: Send + Sync {
 			}
 		};
 
-		let max = ::std::cmp::min(MAX_HEADERS_TO_SEND, req.max);
+		let max = ::std::cmp::min(MAX_HEADERS_PER_REQUEST, req.max);
 
 		let headers: Vec<_> = (0u64..max)
 			.map(|x: u64| x.saturating_mul(req.skip.saturating_add(1)))

rpc/Cargo.toml

@@ -37,6 +37,7 @@ jsonrpc-pubsub = { git = "https://github.com/paritytech/jsonrpc.git", branch = "
 
 ethash = { path = "../ethash" }
 ethcore = { path = "../ethcore", features = ["test-helpers"] }
+fastmap = { path = "../util/fastmap" }
 parity-bytes = { git = "https://github.com/paritytech/parity-common" }
 parity-crypto = { git = "https://github.com/paritytech/parity-common" }
 ethcore-devtools = { path = "../devtools" }

rpc/src/lib.rs

@@ -44,6 +44,7 @@ extern crate jsonrpc_pubsub;
 
 extern crate ethash;
 extern crate ethcore;
+extern crate fastmap;
 extern crate parity_bytes as bytes;
 extern crate parity_crypto as crypto;
 extern crate ethcore_devtools as devtools;

rpc/src/v1/helpers/errors.rs

@@ -101,6 +101,14 @@ pub fn request_rejected_limit() -> Error {
 	}
 }
 
+pub fn request_rejected_param_limit(limit: u64, items_desc: &str) -> Error {
+	Error {
+		code: ErrorCode::ServerError(codes::REQUEST_REJECTED_LIMIT),
+		message: format!("Requested data size exceeds limit of {} {}.", limit, items_desc),
+		data: None,
+	}
+}
+
 pub fn account<T: fmt::Debug>(error: &str, details: T) -> Error {
 	Error {
 		code: ErrorCode::ServerError(codes::ACCOUNT_ERROR),