web3-proxy/web3-proxy/src/connections.rs

461 lines
17 KiB
Rust
Raw Normal View History

2022-05-12 02:50:52 +03:00
///! Load balanced communication with a group of web3 providers
2022-05-18 23:18:01 +03:00
use arc_swap::ArcSwap;
2022-05-05 22:07:09 +03:00
use derive_more::From;
use ethers::prelude::H256;
2022-05-18 19:35:06 +03:00
use futures::future::join_all;
2022-05-05 22:07:09 +03:00
use futures::stream::FuturesUnordered;
use futures::StreamExt;
use governor::clock::{QuantaClock, QuantaInstant};
use governor::NotUntil;
2022-05-20 04:26:02 +03:00
use hashbrown::HashMap;
2022-05-21 01:16:15 +03:00
use serde::ser::{SerializeStruct, Serializer};
use serde::Serialize;
2022-05-05 22:07:09 +03:00
use serde_json::value::RawValue;
use std::cmp;
2022-05-20 04:26:02 +03:00
use std::collections::{BTreeMap, BTreeSet};
2022-05-05 22:07:09 +03:00
use std::fmt;
use std::sync::Arc;
2022-05-17 19:23:27 +03:00
use tokio::task;
2022-05-18 19:35:06 +03:00
use tracing::Instrument;
use tracing::{info, info_span, instrument, trace, warn};
2022-05-05 22:07:09 +03:00
use crate::config::Web3ConnectionConfig;
2022-05-06 08:44:30 +03:00
use crate::connection::{ActiveRequestHandle, Web3Connection};
2022-05-05 22:07:09 +03:00
2022-05-21 01:16:15 +03:00
// Serialize so we can print it on our debug endpoint
#[derive(Clone, Default, Serialize)]
2022-05-18 23:18:01 +03:00
struct SyncedConnections {
head_block_num: u64,
head_block_hash: H256,
2022-05-20 04:26:02 +03:00
inner: BTreeSet<usize>,
2022-05-05 22:07:09 +03:00
}
impl fmt::Debug for SyncedConnections {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
// TODO: the default formatter takes forever to write. this is too quiet though
f.debug_struct("SyncedConnections").finish_non_exhaustive()
}
}
impl SyncedConnections {
2022-05-18 23:18:01 +03:00
pub fn get_head_block_hash(&self) -> &H256 {
&self.head_block_hash
2022-05-15 22:28:22 +03:00
}
2022-05-05 22:07:09 +03:00
}
/// A collection of web3 connections. Sends requests either the current best server or all servers.
#[derive(From)]
pub struct Web3Connections {
inner: Vec<Arc<Web3Connection>>,
2022-05-18 23:18:01 +03:00
synced_connections: ArcSwap<SyncedConnections>,
2022-05-05 22:07:09 +03:00
}
2022-05-21 01:16:15 +03:00
impl Serialize for Web3Connections {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
let inner: Vec<&Web3Connection> = self.inner.iter().map(|x| x.as_ref()).collect();
// 3 is the number of fields in the struct.
let mut state = serializer.serialize_struct("Web3Connections", 2)?;
state.serialize_field("rpcs", &inner)?;
state.serialize_field("synced_connections", &**self.synced_connections.load())?;
state.end()
}
}
2022-05-05 22:07:09 +03:00
impl fmt::Debug for Web3Connections {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
// TODO: the default formatter takes forever to write. this is too quiet though
f.debug_struct("Web3Connections")
.field("inner", &self.inner)
.finish_non_exhaustive()
}
}
impl Web3Connections {
2022-05-17 20:15:18 +03:00
// #[instrument(name = "try_new_Web3Connections", skip_all)]
2022-05-05 22:07:09 +03:00
pub async fn try_new(
2022-05-12 21:49:57 +03:00
chain_id: usize,
2022-05-05 22:07:09 +03:00
servers: Vec<Web3ConnectionConfig>,
http_client: Option<reqwest::Client>,
clock: &QuantaClock,
) -> anyhow::Result<Arc<Self>> {
let num_connections = servers.len();
2022-05-16 01:02:14 +03:00
// turn configs into connections
let mut connections = Vec::with_capacity(num_connections);
2022-05-05 22:07:09 +03:00
for server_config in servers.into_iter() {
2022-05-12 21:49:57 +03:00
match server_config
.try_build(clock, chain_id, http_client.clone())
.await
{
2022-05-06 00:38:15 +03:00
Ok(connection) => connections.push(connection),
2022-05-13 00:29:33 +03:00
Err(e) => warn!("Unable to connect to a server! {:?}", e),
2022-05-06 00:38:15 +03:00
}
2022-05-05 22:07:09 +03:00
}
2022-05-16 01:02:14 +03:00
if connections.len() < 2 {
// TODO: less than 3? what should we do here?
return Err(anyhow::anyhow!(
"need at least 2 connections when subscribing to heads!"
));
}
2022-05-15 22:28:22 +03:00
2022-05-20 05:01:02 +03:00
let synced_connections = SyncedConnections::default();
2022-05-16 01:02:14 +03:00
let connections = Arc::new(Self {
inner: connections,
2022-05-18 23:18:01 +03:00
synced_connections: ArcSwap::new(Arc::new(synced_connections)),
2022-05-16 01:02:14 +03:00
});
2022-05-18 19:35:06 +03:00
Ok(connections)
}
pub async fn subscribe_heads(self: &Arc<Self>) {
let (block_sender, block_receiver) = flume::unbounded();
let mut handles = vec![];
2022-05-18 23:18:01 +03:00
for (rpc_id, connection) in self.inner.iter().enumerate() {
2022-05-18 19:35:06 +03:00
// subscribe to new heads in a spawned future
// TODO: channel instead. then we can have one future with write access to a left-right?
let connection = Arc::clone(connection);
let block_sender = block_sender.clone();
// let url = connection.url().to_string();
let handle = task::Builder::default()
.name("subscribe_new_heads")
2022-05-17 19:23:27 +03:00
.spawn(async move {
2022-05-18 19:35:06 +03:00
// loop to automatically reconnect
// TODO: make this cancellable?
// TODO: instead of passing Some(connections), pass Some(channel_sender). Then listen on the receiver below to keep local heads up-to-date
// TODO: proper spann
connection
2022-05-18 23:18:01 +03:00
.subscribe_new_heads(rpc_id, block_sender.clone(), true)
2022-05-18 19:35:06 +03:00
.instrument(tracing::info_span!("url"))
2022-05-17 19:23:27 +03:00
.await
});
2022-05-18 19:35:06 +03:00
handles.push(handle);
2022-05-16 01:02:14 +03:00
}
2022-05-18 19:35:06 +03:00
let connections = Arc::clone(self);
let handle = task::Builder::default()
.name("update_synced_rpcs")
.spawn(async move { connections.update_synced_rpcs(block_receiver).await });
handles.push(handle);
2022-05-20 05:01:02 +03:00
// TODO: do something with join_all's result
2022-05-18 19:35:06 +03:00
join_all(handles).await;
2022-05-05 22:07:09 +03:00
}
2022-05-18 23:18:01 +03:00
pub fn get_head_block_hash(&self) -> H256 {
*self.synced_connections.load().get_head_block_hash()
2022-05-16 01:02:14 +03:00
}
2022-05-05 22:07:09 +03:00
2022-05-12 02:50:52 +03:00
/// Send the same request to all the handles. Returning the fastest successful result.
2022-05-17 03:56:56 +03:00
#[instrument(skip_all)]
2022-05-12 02:50:52 +03:00
pub async fn try_send_parallel_requests(
2022-05-05 22:07:09 +03:00
self: Arc<Self>,
2022-05-12 02:50:52 +03:00
active_request_handles: Vec<ActiveRequestHandle>,
2022-05-05 22:07:09 +03:00
method: String,
2022-05-12 06:40:41 +03:00
params: Option<Box<RawValue>>,
2022-05-06 08:44:30 +03:00
response_sender: flume::Sender<anyhow::Result<Box<RawValue>>>,
2022-05-05 22:07:09 +03:00
) -> anyhow::Result<()> {
2022-05-12 06:54:42 +03:00
// TODO: if only 1 active_request_handles, do self.try_send_request
2022-05-05 22:07:09 +03:00
let mut unordered_futures = FuturesUnordered::new();
2022-05-12 22:44:31 +03:00
for active_request_handle in active_request_handles {
2022-05-05 22:07:09 +03:00
// clone things so we can pass them to a future
let method = method.clone();
let params = params.clone();
let response_sender = response_sender.clone();
2022-05-17 19:23:27 +03:00
let handle = task::Builder::default()
.name("send_request")
.spawn(async move {
let response: Box<RawValue> =
active_request_handle.request(&method, &params).await?;
// send the first good response to a one shot channel. that way we respond quickly
// drop the result because errors are expected after the first send
response_sender
.send_async(Ok(response))
.await
.map_err(Into::into)
});
2022-05-05 22:07:09 +03:00
unordered_futures.push(handle);
}
2022-05-12 02:50:52 +03:00
// TODO: use iterators instead of pushing into a vec?
2022-05-05 22:07:09 +03:00
let mut errs = vec![];
if let Some(x) = unordered_futures.next().await {
match x.unwrap() {
Ok(_) => {}
Err(e) => {
// TODO: better errors
warn!("Got an error sending request: {}", e);
errs.push(e);
}
}
}
// get the first error (if any)
// TODO: why collect multiple errors if we only pop one?
let e = if !errs.is_empty() {
Err(errs.pop().unwrap())
} else {
Err(anyhow::anyhow!("no successful responses"))
};
// send the error to the channel
2022-05-16 08:16:32 +03:00
if response_sender.send_async(e).await.is_ok() {
2022-05-05 22:07:09 +03:00
// if we were able to send an error, then we never sent a success
return Err(anyhow::anyhow!("no successful responses"));
} else {
// if sending the error failed. the other side must be closed (which means we sent a success earlier)
Ok(())
}
}
2022-05-13 20:43:37 +03:00
/// TODO: possible dead lock here. investigate more. probably refactor
2022-05-18 23:18:01 +03:00
/// TODO: move parts of this onto SyncedConnections?
2022-05-20 05:16:48 +03:00
// we don't instrument here because we put a span inside the while loop
2022-05-16 01:02:14 +03:00
async fn update_synced_rpcs(
&self,
2022-05-18 23:18:01 +03:00
block_receiver: flume::Receiver<(u64, H256, usize)>,
) -> anyhow::Result<()> {
2022-05-18 23:18:01 +03:00
let max_connections = self.inner.len();
2022-05-18 23:28:00 +03:00
let mut connection_states: HashMap<usize, (u64, H256)> =
HashMap::with_capacity(max_connections);
2022-05-18 19:35:06 +03:00
2022-05-20 05:01:02 +03:00
let mut pending_synced_connections = SyncedConnections::default();
2022-05-18 23:18:01 +03:00
while let Ok((new_block_num, new_block_hash, rpc_id)) = block_receiver.recv_async().await {
2022-05-20 00:58:21 +03:00
// TODO: span with more in it?
2022-05-19 06:00:54 +03:00
// TODO: make sure i'm doing this span right
2022-05-20 00:58:21 +03:00
// TODO: show the actual rpc url?
let span = info_span!("block_receiver", rpc_id, new_block_num);
2022-05-18 19:35:06 +03:00
let _enter = span.enter();
2022-05-20 00:58:21 +03:00
if new_block_num == 0 {
warn!("rpc is still syncing");
}
2022-05-18 23:18:01 +03:00
connection_states.insert(rpc_id, (new_block_num, new_block_hash));
// TODO: do something to update the synced blocks
2022-05-18 23:28:00 +03:00
match new_block_num.cmp(&pending_synced_connections.head_block_num) {
2022-05-18 23:18:01 +03:00
cmp::Ordering::Greater => {
// the rpc's newest block is the new overall best block
2022-05-20 05:19:35 +03:00
// TODO: if trace, do the full block hash?
info!("new head: {}", new_block_hash);
2022-05-18 23:18:01 +03:00
2022-05-18 23:28:00 +03:00
pending_synced_connections.inner.clear();
2022-05-19 06:00:54 +03:00
pending_synced_connections.inner.insert(rpc_id);
2022-05-18 23:18:01 +03:00
2022-05-18 23:28:00 +03:00
pending_synced_connections.head_block_num = new_block_num;
2022-05-19 06:00:54 +03:00
// TODO: if the parent hash isn't our previous best block, ignore it
2022-05-18 23:28:00 +03:00
pending_synced_connections.head_block_hash = new_block_hash;
2022-05-18 23:18:01 +03:00
}
cmp::Ordering::Equal => {
2022-05-19 06:00:54 +03:00
if new_block_hash == pending_synced_connections.head_block_hash {
// this rpc has caught up with the best known head
// do not clear synced_connections.
// we just want to add this rpc to the end
// TODO: HashSet here? i think we get dupes if we don't
pending_synced_connections.inner.insert(rpc_id);
} else {
2022-05-18 23:18:01 +03:00
// same height, but different chain
2022-05-19 06:00:54 +03:00
// check connection_states to see which head block is more popular!
let mut rpc_ids_by_block: BTreeMap<H256, Vec<usize>> = BTreeMap::new();
let mut synced_rpcs = 0;
for (rpc_id, (block_num, block_hash)) in connection_states.iter() {
if *block_num != new_block_num {
// this connection isn't synced. we don't care what hash it has
continue;
}
synced_rpcs += 1;
let count = rpc_ids_by_block
.entry(*block_hash)
.or_insert_with(|| Vec::with_capacity(max_connections - 1));
count.push(*rpc_id);
}
let most_common_head_hash = rpc_ids_by_block
.iter()
.max_by(|a, b| a.1.len().cmp(&b.1.len()))
.map(|(k, _v)| k)
.unwrap();
2022-05-18 23:18:01 +03:00
warn!(
2022-05-19 06:00:54 +03:00
"chain is forked! {} possible heads. {}/{}/{} rpcs have {}",
rpc_ids_by_block.len(),
rpc_ids_by_block.get(most_common_head_hash).unwrap().len(),
synced_rpcs,
max_connections,
most_common_head_hash
2022-05-18 23:18:01 +03:00
);
2022-05-05 22:07:09 +03:00
2022-05-19 06:00:54 +03:00
// this isn't the best block in the tier. don't do anything
if !pending_synced_connections.inner.remove(&rpc_id) {
// we didn't remove anything. nothing more to do
continue;
}
// we removed. don't continue so that we update self.synced_connections
}
2022-05-18 23:18:01 +03:00
}
cmp::Ordering::Less => {
// this isn't the best block in the tier. don't do anything
2022-05-19 06:00:54 +03:00
if !pending_synced_connections.inner.remove(&rpc_id) {
// we didn't remove anything. nothing more to do
continue;
}
// we removed. don't continue so that we update self.synced_connections
2022-05-18 23:18:01 +03:00
}
}
// the synced connections have changed
2022-05-18 23:28:00 +03:00
let synced_connections = Arc::new(pending_synced_connections.clone());
2022-05-18 23:18:01 +03:00
2022-05-20 05:11:50 +03:00
trace!(
"rpcs at {}: {:?}",
synced_connections.head_block_hash,
synced_connections.inner
);
// TODO: only publish if there are x (default 2) nodes synced to this block?
2022-05-18 23:18:01 +03:00
// do the arcswap
2022-05-18 23:28:00 +03:00
self.synced_connections.swap(synced_connections);
2022-05-06 08:44:30 +03:00
}
2022-05-05 22:07:09 +03:00
2022-05-18 19:35:06 +03:00
// TODO: if there was an error, we should return it
warn!("block_receiver exited!");
2022-05-05 22:07:09 +03:00
Ok(())
}
/// get the best available rpc server
2022-05-17 03:56:56 +03:00
#[instrument(skip_all)]
2022-05-05 22:07:09 +03:00
pub async fn next_upstream_server(
&self,
) -> Result<ActiveRequestHandle, Option<NotUntil<QuantaInstant>>> {
2022-05-05 22:07:09 +03:00
let mut earliest_not_until = None;
2022-05-19 06:00:54 +03:00
let mut synced_rpc_ids: Vec<usize> = self
.synced_connections
.load()
.inner
.iter()
.cloned()
.collect();
2022-05-05 22:07:09 +03:00
2022-05-19 06:00:54 +03:00
let sort_cache: HashMap<usize, (f32, u32)> = synced_rpc_ids
2022-05-05 22:07:09 +03:00
.iter()
2022-05-18 23:18:01 +03:00
.map(|rpc_id| {
let rpc = self.inner.get(*rpc_id).unwrap();
let active_requests = rpc.active_requests();
let soft_limit = rpc.soft_limit();
2022-05-06 23:44:12 +03:00
let utilization = active_requests as f32 / soft_limit as f32;
2022-05-18 23:38:56 +03:00
(*rpc_id, (utilization, soft_limit))
2022-05-05 22:07:09 +03:00
})
.collect();
2022-05-19 06:00:54 +03:00
synced_rpc_ids.sort_unstable_by(|a, b| {
2022-05-18 23:38:56 +03:00
let (a_utilization, a_soft_limit) = sort_cache.get(a).unwrap();
let (b_utilization, b_soft_limit) = sort_cache.get(b).unwrap();
2022-05-06 23:44:12 +03:00
// TODO: i'm comparing floats. crap
match a_utilization
.partial_cmp(b_utilization)
.unwrap_or(cmp::Ordering::Equal)
{
cmp::Ordering::Equal => a_soft_limit.cmp(b_soft_limit),
x => x,
}
2022-05-05 22:07:09 +03:00
});
2022-05-19 06:00:54 +03:00
// now that the rpcs are sorted, try to get an active request handle for one of them
for rpc_id in synced_rpc_ids.into_iter() {
2022-05-18 23:18:01 +03:00
let rpc = self.inner.get(rpc_id).unwrap();
2022-05-05 22:07:09 +03:00
// increment our connection counter
2022-05-18 23:18:01 +03:00
match rpc.try_request_handle() {
Err(not_until) => {
earliest_possible(&mut earliest_not_until, not_until);
}
2022-05-06 23:44:12 +03:00
Ok(handle) => {
2022-05-18 23:18:01 +03:00
trace!("next server on {:?}: {:?}", self, rpc_id);
2022-05-06 23:44:12 +03:00
return Ok(handle);
}
2022-05-05 22:07:09 +03:00
}
}
2022-05-19 06:00:54 +03:00
warn!("no servers on {:?}! {:?}", self, earliest_not_until);
2022-05-06 23:44:12 +03:00
2022-05-05 22:07:09 +03:00
// this might be None
Err(earliest_not_until)
}
/// get all rpc servers that are not rate limited
2022-05-13 23:50:11 +03:00
/// returns servers even if they aren't in sync. This is useful for broadcasting signed transactions
2022-05-05 22:07:09 +03:00
pub fn get_upstream_servers(
&self,
) -> Result<Vec<ActiveRequestHandle>, Option<NotUntil<QuantaInstant>>> {
2022-05-05 22:07:09 +03:00
let mut earliest_not_until = None;
// TODO: with capacity?
let mut selected_rpcs = vec![];
for connection in self.inner.iter() {
// check rate limits and increment our connection counter
match connection.try_request_handle() {
Err(not_until) => {
earliest_possible(&mut earliest_not_until, not_until);
// this rpc is not available. skip it
}
Ok(handle) => selected_rpcs.push(handle),
2022-05-05 22:07:09 +03:00
}
}
if !selected_rpcs.is_empty() {
return Ok(selected_rpcs);
}
// return the earliest not_until (if no rpcs are synced, this will be None)
Err(earliest_not_until)
}
}
fn earliest_possible(
earliest_not_until_option: &mut Option<NotUntil<QuantaInstant>>,
new_not_until: NotUntil<QuantaInstant>,
) {
match earliest_not_until_option.as_ref() {
None => *earliest_not_until_option = Some(new_not_until),
Some(earliest_not_until) => {
let earliest_possible = earliest_not_until.earliest_possible();
let new_earliest_possible = new_not_until.earliest_possible();
if earliest_possible > new_earliest_possible {
*earliest_not_until_option = Some(new_not_until);
}
}
}
}