gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-05-08 01:05:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

Port eth1 lib to use stable futures

Browse Source
This commit is contained in:
pawan
2020-02-29 18:49:51 +05:30
committed by Age Manning
parent 0b2b379f14
commit a984772475
4 changed files with 3140 additions and 2846 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5214
Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

6
beacon_node/eth1/Cargo.toml
View File

@@ -11,8 +11,8 @@ toml = "^0.5"
web3 = "0.8.0" web3 = "0.8.0"
[dependencies] [dependencies]
reqwest = "0.9" reqwest = "0.10"
futures = "0.1.25" futures = "0.3"
serde_json = "1.0" serde_json = "1.0"
serde = { version = "1.0", features = ["derive"] } serde = { version = "1.0", features = ["derive"] }
hex = "0.3" hex = "0.3"
@@ -24,7 +24,7 @@ tree_hash = { path = "../../eth2/utils/tree_hash"}
eth2_hashing = { path = "../../eth2/utils/eth2_hashing"} eth2_hashing = { path = "../../eth2/utils/eth2_hashing"}
parking_lot = "0.7" parking_lot = "0.7"
slog = "^2.2.3" slog = "^2.2.3"
tokio = "0.1.22" tokio = { version = "0.2", features = ["full"] }
state_processing = { path = "../../eth2/state_processing" } state_processing = { path = "../../eth2/state_processing" }
libflate = "0.1" libflate = "0.1"
lighthouse_metrics = { path = "../../eth2/utils/lighthouse_metrics"} lighthouse_metrics = { path = "../../eth2/utils/lighthouse_metrics"}

277
beacon_node/eth1/src/http.rs
View File

@@ -10,8 +10,8 @@
//! //!
//! There is no ABI parsing here, all function signatures and topics are hard-coded as constants. //! There is no ABI parsing here, all function signatures and topics are hard-coded as constants.
use futures::{Future, Stream}; use futures::future::TryFutureExt;
use reqwest::{header::CONTENT_TYPE, r#async::ClientBuilder, StatusCode}; use reqwest::{header::CONTENT_TYPE, ClientBuilder, StatusCode};
use serde_json::{json, Value}; use serde_json::{json, Value};
use std::ops::Range; use std::ops::Range;
use std::time::Duration; use std::time::Duration;
@@ -40,80 +40,73 @@ pub struct Block {
/// Returns the current block number. /// Returns the current block number.
/// ///
/// Uses HTTP JSON RPC at `endpoint`. E.g., `http://localhost:8545`. /// Uses HTTP JSON RPC at `endpoint`. E.g., `http://localhost:8545`.
pub fn get_block_number( pub async fn get_block_number(endpoint: &str, timeout: Duration) -> Result<u64, String> {
endpoint: &str, let response_body = send_rpc_request(endpoint, "eth_blockNumber", json!([]), timeout).await?;
timeout: Duration, hex_to_u64_be(
) -> impl Future<Item = u64, Error = String> { response_result(&response_body)?
send_rpc_request(endpoint, "eth_blockNumber", json!([]), timeout) .ok_or_else(|| "No result field was returned for block number".to_string())?
.and_then(|response_body| { .as_str()
hex_to_u64_be( .ok_or_else(|| "Data was not string")?,
response_result(&response_body)? )
.ok_or_else(|| "No result field was returned for block number".to_string())? .map_err(|e| format!("Failed to get block number: {}", e))
.as_str()
.ok_or_else(|| "Data was not string")?,
)
})
.map_err(|e| format!("Failed to get block number: {}", e))
} }
/// Gets a block hash by block number. /// Gets a block hash by block number.
/// ///
/// Uses HTTP JSON RPC at `endpoint`. E.g., `http://localhost:8545`. /// Uses HTTP JSON RPC at `endpoint`. E.g., `http://localhost:8545`.
pub fn get_block( pub async fn get_block(
endpoint: &str, endpoint: &str,
block_number: u64, block_number: u64,
timeout: Duration, timeout: Duration,
) -> impl Future<Item = Block, Error = String> { ) -> Result<Block, String> {
let params = json!([ let params = json!([
format!("0x{:x}", block_number), format!("0x{:x}", block_number),
false // do not return full tx objects. false // do not return full tx objects.
]); ]);
send_rpc_request(endpoint, "eth_getBlockByNumber", params, timeout) let response_body = send_rpc_request(endpoint, "eth_getBlockByNumber", params, timeout).await?;
.and_then(|response_body| { let hash = hex_to_bytes(
let hash = hex_to_bytes( response_result(&response_body)?
response_result(&response_body)? .ok_or_else(|| "No result field was returned for block".to_string())?
.ok_or_else(|| "No result field was returned for block".to_string())? .get("hash")
.get("hash") .ok_or_else(|| "No hash for block")?
.ok_or_else(|| "No hash for block")? .as_str()
.as_str() .ok_or_else(|| "Block hash was not string")?,
.ok_or_else(|| "Block hash was not string")?, )?;
)?; let hash = if hash.len() == 32 {
let hash = if hash.len() == 32 { Ok(Hash256::from_slice(&hash))
Ok(Hash256::from_slice(&hash)) } else {
} else { Err(format!("Block has was not 32 bytes: {:?}", hash))
Err(format!("Block has was not 32 bytes: {:?}", hash)) }?;
}?;
let timestamp = hex_to_u64_be( let timestamp = hex_to_u64_be(
response_result(&response_body)? response_result(&response_body)?
.ok_or_else(|| "No result field was returned for timestamp".to_string())? .ok_or_else(|| "No result field was returned for timestamp".to_string())?
.get("timestamp") .get("timestamp")
.ok_or_else(|| "No timestamp for block")? .ok_or_else(|| "No timestamp for block")?
.as_str() .as_str()
.ok_or_else(|| "Block timestamp was not string")?, .ok_or_else(|| "Block timestamp was not string")?,
)?; )?;
let number = hex_to_u64_be( let number = hex_to_u64_be(
response_result(&response_body)? response_result(&response_body)?
.ok_or_else(|| "No result field was returned for number".to_string())? .ok_or_else(|| "No result field was returned for number".to_string())?
.get("number") .get("number")
.ok_or_else(|| "No number for block")? .ok_or_else(|| "No number for block")?
.as_str() .as_str()
.ok_or_else(|| "Block number was not string")?, .ok_or_else(|| "Block number was not string")?,
)?; )?;
if number <= usize::max_value() as u64 { if number <= usize::max_value() as u64 {
Ok(Block { Ok(Block {
hash, hash,
timestamp, timestamp,
number, number,
})
} else {
Err(format!("Block number {} is larger than a usize", number))
}
}) })
.map_err(|e| format!("Failed to get block number: {}", e)) } else {
Err(format!("Block number {} is larger than a usize", number))
}
.map_err(|e| format!("Failed to get block number: {}", e))
} }
/// Returns the value of the `get_deposit_count()` call at the given `address` for the given /// Returns the value of the `get_deposit_count()` call at the given `address` for the given
@@ -122,20 +115,21 @@ pub fn get_block(
/// Assumes that the `address` has the same ABI as the eth2 deposit contract. /// Assumes that the `address` has the same ABI as the eth2 deposit contract.
/// ///
/// Uses HTTP JSON RPC at `endpoint`. E.g., `http://localhost:8545`. /// Uses HTTP JSON RPC at `endpoint`. E.g., `http://localhost:8545`.
pub fn get_deposit_count( pub async fn get_deposit_count(
endpoint: &str, endpoint: &str,
address: &str, address: &str,
block_number: u64, block_number: u64,
timeout: Duration, timeout: Duration,
) -> impl Future<Item = Option<u64>, Error = String> { ) -> Result<Option<u64>, String> {
call( let result = call(
endpoint, endpoint,
address, address,
DEPOSIT_COUNT_FN_SIGNATURE, DEPOSIT_COUNT_FN_SIGNATURE,
block_number, block_number,
timeout, timeout,
) )
.and_then(|result| match result { .await?;
match result {
None => Err("Deposit root response was none".to_string()), None => Err("Deposit root response was none".to_string()),
Some(bytes) => { Some(bytes) => {
if bytes.is_empty() { if bytes.is_empty() {
@@ -151,7 +145,7 @@ pub fn get_deposit_count(
)) ))
} }
} }
}) }
} }
/// Returns the value of the `get_hash_tree_root()` call at the given `block_number`. /// Returns the value of the `get_hash_tree_root()` call at the given `block_number`.
@@ -159,20 +153,21 @@ pub fn get_deposit_count(
/// Assumes that the `address` has the same ABI as the eth2 deposit contract. /// Assumes that the `address` has the same ABI as the eth2 deposit contract.
/// ///
/// Uses HTTP JSON RPC at `endpoint`. E.g., `http://localhost:8545`. /// Uses HTTP JSON RPC at `endpoint`. E.g., `http://localhost:8545`.
pub fn get_deposit_root( pub async fn get_deposit_root(
endpoint: &str, endpoint: &str,
address: &str, address: &str,
block_number: u64, block_number: u64,
timeout: Duration, timeout: Duration,
) -> impl Future<Item = Option<Hash256>, Error = String> { ) -> Result<Option<Hash256>, String> {
call( let result = call(
endpoint, endpoint,
address, address,
DEPOSIT_ROOT_FN_SIGNATURE, DEPOSIT_ROOT_FN_SIGNATURE,
block_number, block_number,
timeout, timeout,
) )
.and_then(|result| match result { .await?;
match result {
None => Err("Deposit root response was none".to_string()), None => Err("Deposit root response was none".to_string()),
Some(bytes) => { Some(bytes) => {
if bytes.is_empty() { if bytes.is_empty() {
@@ -186,7 +181,7 @@ pub fn get_deposit_root(
)) ))
} }
} }
}) }
} }
/// Performs a instant, no-transaction call to the contract `address` with the given `0x`-prefixed /// Performs a instant, no-transaction call to the contract `address` with the given `0x`-prefixed
@@ -195,13 +190,13 @@ pub fn get_deposit_root(
/// Returns bytes, if any. /// Returns bytes, if any.
/// ///
/// Uses HTTP JSON RPC at `endpoint`. E.g., `http://localhost:8545`. /// Uses HTTP JSON RPC at `endpoint`. E.g., `http://localhost:8545`.
fn call( async fn call(
endpoint: &str, endpoint: &str,
address: &str, address: &str,
hex_data: &str, hex_data: &str,
block_number: u64, block_number: u64,
timeout: Duration, timeout: Duration,
) -> impl Future<Item = Option<Vec<u8>>, Error = String> { ) -> Result<Option<Vec<u8>>, String> {
let params = json! ([ let params = json! ([
{ {
"to": address, "to": address,
@@ -210,19 +205,18 @@ fn call(
format!("0x{:x}", block_number) format!("0x{:x}", block_number)
]); ]);
send_rpc_request(endpoint, "eth_call", params, timeout).and_then(|response_body| { let response_body = send_rpc_request(endpoint, "eth_call", params, timeout).await?;
match response_result(&response_body)? { match response_result(&response_body)? {
None => Ok(None), None => Ok(None),
Some(result) => { Some(result) => {
let hex = result let hex = result
.as_str() .as_str()
.map(|s| s.to_string()) .map(|s| s.to_string())
.ok_or_else(|| "'result' value was not a string".to_string())?; .ok_or_else(|| "'result' value was not a string".to_string())?;
Ok(Some(hex_to_bytes(&hex)?)) Ok(Some(hex_to_bytes(&hex)?))
}
} }
}) }
} }
/// A reduced set of fields from an Eth1 contract log. /// A reduced set of fields from an Eth1 contract log.
@@ -238,12 +232,12 @@ pub struct Log {
/// It's not clear from the Ethereum JSON-RPC docs if this range is inclusive or not. /// It's not clear from the Ethereum JSON-RPC docs if this range is inclusive or not.
/// ///
/// Uses HTTP JSON RPC at `endpoint`. E.g., `http://localhost:8545`. /// Uses HTTP JSON RPC at `endpoint`. E.g., `http://localhost:8545`.
pub fn get_deposit_logs_in_range( pub async fn get_deposit_logs_in_range(
endpoint: &str, endpoint: &str,
address: &str, address: &str,
block_height_range: Range<u64>, block_height_range: Range<u64>,
timeout: Duration, timeout: Duration,
) -> impl Future<Item = Vec<Log>, Error = String> { ) -> Result<Vec<Log>, String> {
let params = json! ([{ let params = json! ([{
"address": address, "address": address,
"topics": [DEPOSIT_EVENT_TOPIC], "topics": [DEPOSIT_EVENT_TOPIC],
@@ -251,46 +245,44 @@ pub fn get_deposit_logs_in_range(
"toBlock": format!("0x{:x}", block_height_range.end), "toBlock": format!("0x{:x}", block_height_range.end),
}]); }]);
send_rpc_request(endpoint, "eth_getLogs", params, timeout) let response_body = send_rpc_request(endpoint, "eth_getLogs", params, timeout).await?;
.and_then(|response_body| { response_result(&response_body)?
response_result(&response_body)? .ok_or_else(|| "No result field was returned for deposit logs".to_string())?
.ok_or_else(|| "No result field was returned for deposit logs".to_string())? .as_array()
.as_array() .cloned()
.cloned() .ok_or_else(|| "'result' value was not an array".to_string())?
.ok_or_else(|| "'result' value was not an array".to_string())? .into_iter()
.into_iter() .map(|value| {
.map(|value| { let block_number = value
let block_number = value .get("blockNumber")
.get("blockNumber") .ok_or_else(|| "No block number field in log")?
.ok_or_else(|| "No block number field in log")? .as_str()
.as_str() .ok_or_else(|| "Block number was not string")?;
.ok_or_else(|| "Block number was not string")?;
let data = value let data = value
.get("data") .get("data")
.ok_or_else(|| "No block number field in log")? .ok_or_else(|| "No block number field in log")?
.as_str() .as_str()
.ok_or_else(|| "Data was not string")?; .ok_or_else(|| "Data was not string")?;
Ok(Log { Ok(Log {
block_number: hex_to_u64_be(&block_number)?, block_number: hex_to_u64_be(&block_number)?,
data: hex_to_bytes(data)?, data: hex_to_bytes(data)?,
}) })
})
.collect::<Result<Vec<Log>, String>>()
}) })
.collect::<Result<Vec<Log>, String>>()
.map_err(|e| format!("Failed to get logs in range: {}", e)) .map_err(|e| format!("Failed to get logs in range: {}", e))
} }
/// Sends an RPC request to `endpoint`, using a POST with the given `body`. /// Sends an RPC request to `endpoint`, using a POST with the given `body`.
/// ///
/// Tries to receive the response and parse the body as a `String`. /// Tries to receive the response and parse the body as a `String`.
pub fn send_rpc_request( pub async fn send_rpc_request(
endpoint: &str, endpoint: &str,
method: &str, method: &str,
params: Value, params: Value,
timeout: Duration, timeout: Duration,
) -> impl Future<Item = String, Error = String> { ) -> Result<String, String> {
let body = json! ({ let body = json! ({
"jsonrpc": "2.0", "jsonrpc": "2.0",
"method": method, "method": method,
@@ -303,7 +295,7 @@ pub fn send_rpc_request(
// //
// A better solution would be to create some struct that contains a built client and pass it // A better solution would be to create some struct that contains a built client and pass it
// around (similar to the `web3` crate's `Transport` structs). // around (similar to the `web3` crate's `Transport` structs).
ClientBuilder::new() let response = ClientBuilder::new()
.timeout(timeout) .timeout(timeout)
.build() .build()
.expect("The builder should always build a client") .expect("The builder should always build a client")
@@ -312,43 +304,32 @@ pub fn send_rpc_request(
.body(body) .body(body)
.send() .send()
.map_err(|e| format!("Request failed: {:?}", e)) .map_err(|e| format!("Request failed: {:?}", e))
.and_then(|response| { .await?;
if response.status() != StatusCode::OK { if response.status() != StatusCode::OK {
Err(format!( return Err(format!(
"Response HTTP status was not 200 OK: {}.", "Response HTTP status was not 200 OK: {}.",
response.status() response.status()
)) ));
} else { };
Ok(response) let encoding = response
} .headers()
}) .get(CONTENT_TYPE)
.and_then(|response| { .ok_or_else(|| "No content-type header in response".to_string())?
response .to_str()
.headers() .map(|s| s.to_string())
.get(CONTENT_TYPE) .map_err(|e| format!("Failed to parse content-type header: {}", e))?;
.ok_or_else(|| "No content-type header in response".to_string())
.and_then(|encoding| { response
encoding .bytes()
.to_str() .map_err(|e| format!("Failed to receive body: {:?}", e))
.map(|s| s.to_string()) .await
.map_err(|e| format!("Failed to parse content-type header: {}", e)) .and_then(move |bytes| match encoding.as_str() {
}) "application/json" => Ok(bytes),
.map(|encoding| (response, encoding)) "application/json; charset=utf-8" => Ok(bytes),
}) other => Err(format!("Unsupported encoding: {}", other)),
.and_then(|(response, encoding)| {
response
.into_body()
.concat2()
.map(|chunk| chunk.iter().cloned().collect::<Vec<u8>>())
.map_err(|e| format!("Failed to receive body: {:?}", e))
.and_then(move |bytes| match encoding.as_str() {
"application/json" => Ok(bytes),
"application/json; charset=utf-8" => Ok(bytes),
other => Err(format!("Unsupported encoding: {}", other)),
})
.map(|bytes| String::from_utf8_lossy(&bytes).into_owned())
.map_err(|e| format!("Failed to receive body: {:?}", e))
}) })
.map(|bytes| String::from_utf8_lossy(&bytes).into_owned())
.map_err(|e| format!("Failed to receive body: {:?}", e))
} }
/// Accepts an entire HTTP body (as a string) and returns the `result` field, as a serde `Value`. /// Accepts an entire HTTP body (as a string) and returns the `result` field, as a serde `Value`.

489
beacon_node/eth1/src/service.rs
View File

@@ -7,16 +7,18 @@ use crate::{
DepositLog, DepositLog,
}; };
use futures::{ use futures::{
future::{loop_fn, Loop}, future::{FutureExt, TryFutureExt},
stream, Future, Stream, stream,
stream::TryStreamExt,
}; };
use parking_lot::{RwLock, RwLockReadGuard}; use parking_lot::{RwLock, RwLockReadGuard};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use slog::{debug, error, trace, Logger}; use slog::{debug, error, trace, Logger};
use std::ops::{Range, RangeInclusive}; use std::ops::{Range, RangeInclusive};
use std::sync::Arc; use std::sync::Arc;
use std::time::{Duration, Instant}; use std::time::Duration;
use tokio::timer::Delay; use tokio::sync::oneshot::error::TryRecvError;
use tokio::time::delay_for;
const STANDARD_TIMEOUT_MILLIS: u64 = 15_000; const STANDARD_TIMEOUT_MILLIS: u64 = 15_000;
@@ -245,29 +247,23 @@ impl Service {
/// - Err(_) if there is an error. /// - Err(_) if there is an error.
/// ///
/// Emits logs for debugging and errors. /// Emits logs for debugging and errors.
pub fn update( pub async fn update(
&self, &self,
) -> impl Future<Item = (DepositCacheUpdateOutcome, BlockCacheUpdateOutcome), Error = String> ) -> Result<(DepositCacheUpdateOutcome, BlockCacheUpdateOutcome), String> {
{
let log_a = self.log.clone();
let log_b = self.log.clone();
let inner_1 = self.inner.clone();
let inner_2 = self.inner.clone();
let deposit_future = self let deposit_future = self
.update_deposit_cache() .update_deposit_cache()
.map_err(|e| format!("Failed to update eth1 cache: {:?}", e)) .map_err(|e| format!("Failed to update eth1 cache: {:?}", e))
.then(move |result| { .then(|result| async move{
match &result { match &result {
Ok(DepositCacheUpdateOutcome::Success { logs_imported }) => trace!( Ok(DepositCacheUpdateOutcome::Success { logs_imported }) => trace!(
log_a, self.log,
"Updated eth1 deposit cache"; "Updated eth1 deposit cache";
"cached_deposits" => inner_1.deposit_cache.read().cache.len(), "cached_deposits" => self.inner.deposit_cache.read().cache.len(),
"logs_imported" => logs_imported, "logs_imported" => logs_imported,
"last_processed_eth1_block" => inner_1.deposit_cache.read().last_processed_block, "last_processed_eth1_block" => self.inner.deposit_cache.read().last_processed_block,
), ),
Err(e) => error!( Err(e) => error!(
log_a, self.log,
"Failed to update eth1 deposit cache"; "Failed to update eth1 deposit cache";
"error" => e "error" => e
), ),
@@ -279,20 +275,20 @@ impl Service {
let block_future = self let block_future = self
.update_block_cache() .update_block_cache()
.map_err(|e| format!("Failed to update eth1 cache: {:?}", e)) .map_err(|e| format!("Failed to update eth1 cache: {:?}", e))
.then(move |result| { .then(|result| async move {
match &result { match &result {
Ok(BlockCacheUpdateOutcome::Success { Ok(BlockCacheUpdateOutcome::Success {
blocks_imported, blocks_imported,
head_block_number, head_block_number,
}) => trace!( }) => trace!(
log_b, self.log,
"Updated eth1 block cache"; "Updated eth1 block cache";
"cached_blocks" => inner_2.block_cache.read().len(), "cached_blocks" => self.inner.block_cache.read().len(),
"blocks_imported" => blocks_imported, "blocks_imported" => blocks_imported,
"head_block" => head_block_number, "head_block" => head_block_number,
), ),
Err(e) => error!( Err(e) => error!(
log_b, self.log,
"Failed to update eth1 block cache"; "Failed to update eth1 block cache";
"error" => e "error" => e
), ),
@@ -301,7 +297,7 @@ impl Service {
result result
}); });
deposit_future.join(block_future) futures::try_join!(deposit_future, block_future)
} }
/// A looping future that updates the cache, then waits `config.auto_update_interval` before /// A looping future that updates the cache, then waits `config.auto_update_interval` before
@@ -313,56 +309,40 @@ impl Service {
/// - Err(_) if there is an error. /// - Err(_) if there is an error.
/// ///
/// Emits logs for debugging and errors. /// Emits logs for debugging and errors.
pub fn auto_update( pub async fn auto_update(
&self, &self,
exit: tokio::sync::oneshot::Receiver<()>, mut exit: tokio::sync::oneshot::Receiver<()>,
) -> impl Future<Item = (), Error = ()> { ) -> Result<(), ()> {
let service = self.clone();
let log = self.log.clone();
let update_interval = Duration::from_millis(self.config().auto_update_interval_millis); let update_interval = Duration::from_millis(self.config().auto_update_interval_millis);
let loop_future = loop_fn((), move |()| { loop {
let service = service.clone(); let update_result = self.update().await;
let log_a = log.clone();
let log_b = log.clone();
service match update_result {
.update() Err(e) => error!(
.then(move |update_result| { self.log,
match update_result { "Failed to update eth1 cache";
Err(e) => error!( "retry_millis" => update_interval.as_millis(),
log_a, "error" => e,
"Failed to update eth1 cache"; ),
"retry_millis" => update_interval.as_millis(), Ok((deposit, block)) => debug!(
"error" => e, self.log,
), "Updated eth1 cache";
Ok((deposit, block)) => debug!( "retry_millis" => update_interval.as_millis(),
log_a, "blocks" => format!("{:?}", block),
"Updated eth1 cache"; "deposits" => format!("{:?}", deposit),
"retry_millis" => update_interval.as_millis(), ),
"blocks" => format!("{:?}", block), };
"deposits" => format!("{:?}", deposit),
),
};
// Do not break the loop if there is an update failure. // WARNING: delay_for doesn't return an error and panics on error.
Ok(()) delay_for(update_interval).await;
}) match exit.try_recv() {
.and_then(move |_| Delay::new(Instant::now() + update_interval)) Ok(_) | Err(TryRecvError::Closed) => break,
.then(move |timer_result| { Err(TryRecvError::Empty) => {}
if let Err(e) = timer_result { }
error!( }
log_b,
"Failed to trigger eth1 cache update delay";
"error" => format!("{:?}", e),
);
}
// Do not break the loop if there is an timer failure.
Ok(Loop::Continue(()))
})
});
loop_future.select(exit).map(|_| ()).map_err(|_| ()) Ok(())
} }
/// Contacts the remote eth1 node and attempts to import deposit logs up to the configured /// Contacts the remote eth1 node and attempts to import deposit logs up to the configured
@@ -377,11 +357,7 @@ impl Service {
/// - Err(_) if there is an error. /// - Err(_) if there is an error.
/// ///
/// Emits logs for debugging and errors. /// Emits logs for debugging and errors.
pub fn update_deposit_cache( pub async fn update_deposit_cache(&self) -> Result<DepositCacheUpdateOutcome, Error> {
&self,
) -> impl Future<Item = DepositCacheUpdateOutcome, Error = Error> {
let service_1 = self.clone();
let service_2 = self.clone();
let blocks_per_log_query = self.config().blocks_per_log_query; let blocks_per_log_query = self.config().blocks_per_log_query;
let max_log_requests_per_update = self let max_log_requests_per_update = self
.config() .config()
@@ -395,97 +371,98 @@ impl Service {
.map(|n| n + 1) .map(|n| n + 1)
.unwrap_or_else(|| self.config().deposit_contract_deploy_block); .unwrap_or_else(|| self.config().deposit_contract_deploy_block);
get_new_block_numbers( let range = get_new_block_numbers(
&self.config().endpoint, &self.config().endpoint,
next_required_block, next_required_block,
self.config().follow_distance, self.config().follow_distance,
) )
.map(move |range| { .await?;
let block_number_chunks = if let Some(range) = range {
range range
.map(|range| { .collect::<Vec<u64>>()
range .chunks(blocks_per_log_query)
.collect::<Vec<u64>>() .take(max_log_requests_per_update)
.chunks(blocks_per_log_query) .map(|vec| {
.take(max_log_requests_per_update) let first = vec.first().cloned().unwrap_or_else(|| 0);
.map(|vec| { let last = vec.last().map(|n| n + 1).unwrap_or_else(|| 0);
let first = vec.first().cloned().unwrap_or_else(|| 0); first..last
let last = vec.last().map(|n| n + 1).unwrap_or_else(|| 0);
first..last
})
.collect::<Vec<Range<u64>>>()
}) })
.unwrap_or_else(|| vec![]) .collect::<Vec<Range<u64>>>()
}) } else {
.and_then(move |block_number_chunks| { Vec::new()
stream::unfold( };
block_number_chunks.into_iter(),
move |mut chunks| match chunks.next() { stream::try_unfold(block_number_chunks.into_iter(), |mut chunks| async move {
Some(chunk) => { match chunks.next() {
let chunk_1 = chunk.clone(); Some(chunk) => {
Some( let chunk_1 = chunk.clone();
get_deposit_logs_in_range( match get_deposit_logs_in_range(
&service_1.config().endpoint, &self.config().endpoint,
&service_1.config().deposit_contract_address, &self.config().deposit_contract_address,
chunk, chunk,
Duration::from_millis(GET_DEPOSIT_LOG_TIMEOUT_MILLIS), Duration::from_millis(GET_DEPOSIT_LOG_TIMEOUT_MILLIS),
) )
.map_err(Error::GetDepositLogsFailed) .await
.map(|logs| (chunk_1, logs)) {
.map(|logs| (logs, chunks)), Ok(logs) => Ok(Some(((chunk_1, logs), chunks))),
) Err(e) => Err(Error::GetDepositLogsFailed(e)),
} }
None => None, }
}, None => Ok(None),
) }
.fold(0, move |mut sum, (block_range, log_chunk)| {
let mut cache = service_2.deposits().write();
log_chunk
.into_iter()
.map(|raw_log| {
DepositLog::from_log(&raw_log).map_err(|error| {
Error::FailedToParseDepositLog {
block_range: block_range.clone(),
error,
}
})
})
// Return early if any of the logs cannot be parsed.
//
// This costs an additional `collect`, however it enforces that no logs are
// imported if any one of them cannot be parsed.
.collect::<Result<Vec<_>, _>>()?
.into_iter()
.map(|deposit_log| {
cache
.cache
.insert_log(deposit_log)
.map_err(Error::FailedToInsertDeposit)?;
sum += 1;
Ok(())
})
// Returns if a deposit is unable to be added to the cache.
//
// If this error occurs, the cache will no longer be guaranteed to hold either
// none or all of the logs for each block (i.e., they may exist _some_ logs for
// a block, but not _all_ logs for that block). This scenario can cause the
// node to choose an invalid genesis state or propose an invalid block.
.collect::<Result<_, _>>()?;
cache.last_processed_block = Some(block_range.end.saturating_sub(1));
metrics::set_gauge(&metrics::DEPOSIT_CACHE_LEN, cache.cache.len() as i64);
metrics::set_gauge(
&metrics::HIGHEST_PROCESSED_DEPOSIT_BLOCK,
cache.last_processed_block.unwrap_or_else(|| 0) as i64,
);
Ok(sum)
})
.map(|logs_imported| DepositCacheUpdateOutcome::Success { logs_imported })
}) })
.try_fold(0, |mut sum: usize, (block_range, log_chunk)| async move {
let mut cache = self.deposits().write();
log_chunk
.into_iter()
.map(|raw_log| {
DepositLog::from_log(&raw_log).map_err(|error| Error::FailedToParseDepositLog {
block_range: block_range.clone(),
error,
})
})
// Return early if any of the logs cannot be parsed.
//
// This costs an additional `collect`, however it enforces that no logs are
// imported if any one of them cannot be parsed.
.collect::<Result<Vec<_>, _>>()?
.into_iter()
.map(|deposit_log| {
cache
.cache
.insert_log(deposit_log)
.map_err(Error::FailedToInsertDeposit)?;
sum += 1;
Ok(())
})
// Returns if a deposit is unable to be added to the cache.
//
// If this error occurs, the cache will no longer be guaranteed to hold either
// none or all of the logs for each block (i.e., they may exist _some_ logs for
// a block, but not _all_ logs for that block). This scenario can cause the
// node to choose an invalid genesis state or propose an invalid block.
.collect::<Result<_, _>>()?;
cache.last_processed_block = Some(block_range.end.saturating_sub(1));
metrics::set_gauge(&metrics::DEPOSIT_CACHE_LEN, cache.cache.len() as i64);
metrics::set_gauge(
&metrics::HIGHEST_PROCESSED_DEPOSIT_BLOCK,
cache.last_processed_block.unwrap_or_else(|| 0) as i64,
);
Ok(sum)
})
.map(|logs_imported| {
Ok(DepositCacheUpdateOutcome::Success {
logs_imported: logs_imported?,
})
})
.await
} }
/// Contacts the remote eth1 node and attempts to import all blocks up to the configured /// Contacts the remote eth1 node and attempts to import all blocks up to the configured
@@ -499,90 +476,90 @@ impl Service {
/// - Err(_) if there is an error. /// - Err(_) if there is an error.
/// ///
/// Emits logs for debugging and errors. /// Emits logs for debugging and errors.
pub fn update_block_cache(&self) -> impl Future<Item = BlockCacheUpdateOutcome, Error = Error> { pub async fn update_block_cache(&self) -> Result<BlockCacheUpdateOutcome, Error> {
let cache_1 = self.inner.clone();
let cache_2 = self.inner.clone();
let cache_3 = self.inner.clone();
let cache_4 = self.inner.clone();
let cache_5 = self.inner.clone();
let cache_6 = self.inner.clone();
let block_cache_truncation = self.config().block_cache_truncation; let block_cache_truncation = self.config().block_cache_truncation;
let max_blocks_per_update = self let max_blocks_per_update = self
.config() .config()
.max_blocks_per_update .max_blocks_per_update
.unwrap_or_else(usize::max_value); .unwrap_or_else(usize::max_value);
let next_required_block = cache_1 let next_required_block = self
.inner
.block_cache .block_cache
.read() .read()
.highest_block_number() .highest_block_number()
.map(|n| n + 1) .map(|n| n + 1)
.unwrap_or_else(|| self.config().lowest_cached_block_number); .unwrap_or_else(|| self.config().lowest_cached_block_number);
get_new_block_numbers( let range = get_new_block_numbers(
&self.config().endpoint, &self.config().endpoint,
next_required_block, next_required_block,
self.config().follow_distance, self.config().follow_distance,
) )
.await?;
// Map the range of required blocks into a Vec. // Map the range of required blocks into a Vec.
// //
// If the required range is larger than the size of the cache, drop the exiting cache // If the required range is larger than the size of the cache, drop the exiting cache
// because it's exipred and just download enough blocks to fill the cache. // because it's exipred and just download enough blocks to fill the cache.
.and_then(move |range| { let required_block_numbers = if let Some(range) = range {
range if range.start() > range.end() {
.map(|range| { // Note: this check is not strictly necessary, however it remains to safe
if range.start() > range.end() { // guard against any regression which may cause an underflow in a following
// Note: this check is not strictly necessary, however it remains to safe // subtraction operation.
// guard against any regression which may cause an underflow in a following return Err(Error::Internal("Range was not increasing".into()));
// subtraction operation. } else {
Err(Error::Internal("Range was not increasing".into())) let range_size = range.end() - range.start();
} else { let max_size = block_cache_truncation
let range_size = range.end() - range.start(); .map(|n| n as u64)
let max_size = block_cache_truncation .unwrap_or_else(u64::max_value);
.map(|n| n as u64) if range_size > max_size {
.unwrap_or_else(u64::max_value); // If the range of required blocks is larger than `max_size`, drop all
if range_size > max_size { // existing blocks and download `max_size` count of blocks.
// If the range of required blocks is larger than `max_size`, drop all let first_block = range.end() - max_size;
// existing blocks and download `max_size` count of blocks. (*self.inner.block_cache.write()) = BlockCache::default();
let first_block = range.end() - max_size; (first_block..=*range.end()).collect::<Vec<u64>>()
(*cache_5.block_cache.write()) = BlockCache::default(); } else {
Ok((first_block..=*range.end()).collect::<Vec<u64>>()) range.collect::<Vec<u64>>()
} else { }
Ok(range.collect::<Vec<u64>>()) }
} else {
Vec::new()
};
// Download the range of blocks and sequentially import them into the cache.
// Last processed block in deposit cache
let latest_in_cache = self
.inner
.deposit_cache
.read()
.last_processed_block
.unwrap_or(0);
let required_block_numbers = required_block_numbers
.into_iter()
.filter(|x| *x <= latest_in_cache)
.take(max_blocks_per_update)
.collect::<Vec<_>>();
// Produce a stream from the list of required block numbers and return a future that
// consumes the it.
let eth1_blocks = stream::try_unfold(
required_block_numbers.into_iter(),
|mut block_numbers| async move {
match block_numbers.next() {
Some(block_number) => {
match download_eth1_block(self.inner.clone(), block_number).await {
Ok(eth1_block) => Ok(Some((eth1_block, block_numbers))),
Err(e) => Err(e),
} }
} }
}) None => Ok(None),
.unwrap_or_else(|| Ok(vec![])) }
}) },
// Download the range of blocks and sequentially import them into the cache. );
.and_then(move |required_block_numbers| {
// Last processed block in deposit cache
let latest_in_cache = cache_6
.deposit_cache
.read()
.last_processed_block
.unwrap_or(0);
let required_block_numbers = required_block_numbers let blocks_imported = eth1_blocks
.into_iter() .try_fold(0, |sum: usize, eth1_block| async move {
.filter(|x| *x <= latest_in_cache) self.inner
.take(max_blocks_per_update)
.collect::<Vec<_>>();
// Produce a stream from the list of required block numbers and return a future that
// consumes the it.
stream::unfold(
required_block_numbers.into_iter(),
move |mut block_numbers| match block_numbers.next() {
Some(block_number) => Some(
download_eth1_block(cache_2.clone(), block_number)
.map(|v| (v, block_numbers)),
),
None => None,
},
)
.fold(0, move |sum, eth1_block| {
cache_3
.block_cache .block_cache
.write() .write()
.insert_root_or_child(eth1_block) .insert_root_or_child(eth1_block)
@@ -590,11 +567,11 @@ impl Service {
metrics::set_gauge( metrics::set_gauge(
&metrics::BLOCK_CACHE_LEN, &metrics::BLOCK_CACHE_LEN,
cache_3.block_cache.read().len() as i64, self.inner.block_cache.read().len() as i64,
); );
metrics::set_gauge( metrics::set_gauge(
&metrics::LATEST_CACHED_BLOCK_TIMESTAMP, &metrics::LATEST_CACHED_BLOCK_TIMESTAMP,
cache_3 self.inner
.block_cache .block_cache
.read() .read()
.latest_block_timestamp() .latest_block_timestamp()
@@ -603,64 +580,59 @@ impl Service {
Ok(sum + 1) Ok(sum + 1)
}) })
}) .await?;
.and_then(move |blocks_imported| { // Prune the block cache, preventing it from growing too large.
// Prune the block cache, preventing it from growing too large. self.inner.prune_blocks();
cache_4.prune_blocks();
metrics::set_gauge( metrics::set_gauge(
&metrics::BLOCK_CACHE_LEN, &metrics::BLOCK_CACHE_LEN,
cache_4.block_cache.read().len() as i64, self.inner.block_cache.read().len() as i64,
); );
Ok(BlockCacheUpdateOutcome::Success { Ok(BlockCacheUpdateOutcome::Success {
blocks_imported, blocks_imported,
head_block_number: cache_4.block_cache.read().highest_block_number(), head_block_number: self.inner.block_cache.read().highest_block_number(),
})
}) })
} }
} }
/// Determine the range of blocks that need to be downloaded, given the remotes best block and /// Determine the range of blocks that need to be downloaded, given the remotes best block and
/// the locally stored best block. /// the locally stored best block.
fn get_new_block_numbers<'a>( async fn get_new_block_numbers<'a>(
endpoint: &str, endpoint: &str,
next_required_block: u64, next_required_block: u64,
follow_distance: u64, follow_distance: u64,
) -> impl Future<Item = Option<RangeInclusive<u64>>, Error = Error> + 'a { ) -> Result<Option<RangeInclusive<u64>>, Error> {
get_block_number(endpoint, Duration::from_millis(BLOCK_NUMBER_TIMEOUT_MILLIS)) let remote_highest_block =
.map_err(Error::GetBlockNumberFailed) get_block_number(endpoint, Duration::from_millis(BLOCK_NUMBER_TIMEOUT_MILLIS))
.and_then(move |remote_highest_block| { .map_err(Error::GetBlockNumberFailed)
let remote_follow_block = remote_highest_block.saturating_sub(follow_distance); .await?;
let remote_follow_block = remote_highest_block.saturating_sub(follow_distance);
if next_required_block <= remote_follow_block { if next_required_block <= remote_follow_block {
Ok(Some(next_required_block..=remote_follow_block)) Ok(Some(next_required_block..=remote_follow_block))
} else if next_required_block > remote_highest_block + 1 { } else if next_required_block > remote_highest_block + 1 {
// If this is the case, the node must have gone "backwards" in terms of it's sync // If this is the case, the node must have gone "backwards" in terms of it's sync
// (i.e., it's head block is lower than it was before). // (i.e., it's head block is lower than it was before).
// //
// We assume that the `follow_distance` should be sufficient to ensure this never // We assume that the `follow_distance` should be sufficient to ensure this never
// happens, otherwise it is an error. // happens, otherwise it is an error.
Err(Error::RemoteNotSynced { Err(Error::RemoteNotSynced {
next_required_block, next_required_block,
remote_highest_block, remote_highest_block,
follow_distance, follow_distance,
})
} else {
// Return an empty range.
Ok(None)
}
}) })
} else {
// Return an empty range.
Ok(None)
}
} }
/// Downloads the `(block, deposit_root, deposit_count)` tuple from an eth1 node for the given /// Downloads the `(block, deposit_root, deposit_count)` tuple from an eth1 node for the given
/// `block_number`. /// `block_number`.
/// ///
/// Performs three async calls to an Eth1 HTTP JSON RPC endpoint. /// Performs three async calls to an Eth1 HTTP JSON RPC endpoint.
fn download_eth1_block<'a>( async fn download_eth1_block(cache: Arc<Inner>, block_number: u64) -> Result<Eth1Block, Error> {
cache: Arc<Inner>,
block_number: u64,
) -> impl Future<Item = Eth1Block, Error = Error> + 'a {
let deposit_root = cache let deposit_root = cache
.deposit_cache .deposit_cache
.read() .read()
@@ -672,13 +644,14 @@ fn download_eth1_block<'a>(
.cache .cache
.get_deposit_count_from_cache(block_number); .get_deposit_count_from_cache(block_number);
// Performs a `get_blockByNumber` call to an eth1 node. // Performs a `get_blockByNumber` call to an eth1 node.
get_block( let http_block = get_block(
&cache.config.read().endpoint, &cache.config.read().endpoint,
block_number, block_number,
Duration::from_millis(GET_BLOCK_TIMEOUT_MILLIS), Duration::from_millis(GET_BLOCK_TIMEOUT_MILLIS),
) )
.map_err(Error::BlockDownloadFailed) .map_err(Error::BlockDownloadFailed)
.map(move |http_block| Eth1Block { .await?;
Ok(Eth1Block {
hash: http_block.hash, hash: http_block.hash,
number: http_block.number, number: http_block.number,
timestamp: http_block.timestamp, timestamp: http_block.timestamp,