ethrpc_rs/
evaluator.rs

1//! Server list ([`RpcList`]) and fastest-server selection ([`evaluate`](crate::evaluate)).
2
3use std::time::{Duration, Instant};
4
5use async_trait::async_trait;
6use futures::stream::{FuturesUnordered, StreamExt};
7use serde_json::Value;
8
9use crate::decode::ValueExt;
10use crate::error::{Error, Result};
11use crate::rpc::{Handler, Rpc};
12
13/// A list of [`Rpc`] endpoints that implements [`Handler`] with failover.
14#[derive(Default)]
15pub struct RpcList(pub Vec<Rpc>);
16
17#[async_trait]
18impl Handler for RpcList {
19    /// Performs a call against the servers in order, failing over to the next on
20    /// transport errors. A JSON-RPC error response is returned immediately
21    /// without failover, since it represents a valid answer from the server.
22    async fn call(&self, method: &str, params: Vec<Value>) -> Result<Value> {
23        if self.0.is_empty() {
24            return Err(Error::NoAvailableServer);
25        }
26        let mut last_err: Option<Error> = None;
27        for srv in &self.0 {
28            match srv.call(method, params.clone()).await {
29                Ok(res) => return Ok(res),
30                Err(err) => {
31                    // A JSON-RPC error is a valid response — don't retry.
32                    if err.is_rpc_error() {
33                        return Err(err);
34                    }
35                    last_err = Some(err);
36                }
37            }
38        }
39        Err(last_err.unwrap_or(Error::NoAvailableServer))
40    }
41}
42
43/// Probes a single server with `eth_blockNumber`, recording its lag and block.
44async fn probe(host: String) -> Result<Rpc> {
45    let mut r = Rpc::new(host);
46    let start = Instant::now();
47    let block = r.call("eth_blockNumber", vec![]).await?.to_u64()?;
48    r.set_probe(start.elapsed(), block);
49    Ok(r)
50}
51
52/// How long [`evaluate`](crate::evaluate) keeps waiting for additional servers after the first
53/// success, so it doesn't block on the slowest endpoint.
54const SELECTION_GRACE: Duration = Duration::from_millis(200);
55
56/// Calls every server with `eth_blockNumber`, measures response time, and
57/// returns a [`Handler`] backed by the servers that responded.
58///
59/// With a single server, the returned handler is that one [`Rpc`] (and an error
60/// is returned if it fails to respond). With multiple servers, the result is an
61/// [`RpcList`] of every server that answered within a short grace period
62/// (200&nbsp;ms) of the first success (or all that eventually answer, whichever
63/// comes first).
64pub async fn evaluate(servers: &[&str]) -> Result<Box<dyn Handler>> {
65    match servers.len() {
66        0 => Err(Error::NoAvailableServer),
67        1 => {
68            // Probe it so we honor the contract of returning working servers.
69            let r = probe(servers[0].to_string()).await?;
70            Ok(Box::new(r))
71        }
72        _ => {
73            let mut futs: FuturesUnordered<_> =
74                servers.iter().map(|s| probe(s.to_string())).collect();
75
76            let mut res: Vec<Rpc> = Vec::new();
77            let mut last_err: Option<Error> = None;
78            // Armed after the first success; once it fires we stop waiting.
79            let mut grace = std::pin::pin!(futures::future::OptionFuture::from(None));
80
81            loop {
82                tokio::select! {
83                    biased;
84                    _ = grace.as_mut(), if !res.is_empty() => {
85                        return Ok(Box::new(RpcList(res)));
86                    }
87                    next = futs.next() => match next {
88                        None => {
89                            // Every server has reported in.
90                            if !res.is_empty() {
91                                return Ok(Box::new(RpcList(res)));
92                            }
93                            return Err(last_err.unwrap_or(Error::NoAvailableServer));
94                        }
95                        Some(Ok(r)) => {
96                            let first = res.is_empty();
97                            res.push(r);
98                            if first {
99                                grace.set(Some(tokio::time::sleep(SELECTION_GRACE)).into());
100                            }
101                        }
102                        Some(Err(e)) => {
103                            last_err = Some(e);
104                        }
105                    }
106                }
107            }
108        }
109    }
110}
ethrpc_rs/evaluator.rs

ethrpc_rs/
evaluator.rs
