tornadocontrib/web3-proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

move timeout deeper

Browse Source
This commit is contained in:
Bryan Stitt 2023-05-23 15:50:29 -07:00
parent 55f9c57827
commit dafb69fae1
1 changed files with 37 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
web3_proxy/src/app/mod.rs
View File

@ -1011,20 +1011,12 @@ impl Web3ProxyApp {
) -> Web3ProxyResult<(StatusCode, JsonRpcForwardedResponseEnum, Vec<Arc<Web3Rpc>>)> { ) -> Web3ProxyResult<(StatusCode, JsonRpcForwardedResponseEnum, Vec<Arc<Web3Rpc>>)> {
// trace!(?request, "proxy_web3_rpc"); // trace!(?request, "proxy_web3_rpc");
// even though we have timeouts on the requests to our backend providers,
// we need a timeout for the incoming request so that retries don't run forever
// TODO: take this as an optional argument. check for a different max from the user_tier?
// TODO: how much time was spent on this request alredy?
let max_time = Duration::from_secs(240);
// TODO: use streams and buffers so we don't overwhelm our server // TODO: use streams and buffers so we don't overwhelm our server
let response = match request { let response = match request {
JsonRpcRequestEnum::Single(mut request) => { JsonRpcRequestEnum::Single(mut request) => {
let (status_code, response, rpcs) = timeout( let (status_code, response, rpcs) = self
max_time, .proxy_cached_request(&authorization, &mut request, None)
self.proxy_cached_request(&authorization, &mut request, None), .await;
)
.await?;
( (
status_code, status_code,
@ -1033,11 +1025,9 @@ impl Web3ProxyApp {
) )
} }
JsonRpcRequestEnum::Batch(requests) => { JsonRpcRequestEnum::Batch(requests) => {
let (responses, rpcs) = timeout( let (responses, rpcs) = self
max_time, .proxy_web3_rpc_requests(&authorization, requests)
self.proxy_web3_rpc_requests(&authorization, requests), .await?;
)
.await??;
// TODO: real status code // TODO: real status code
( (
@ -1331,6 +1321,7 @@ impl Web3ProxyApp {
| "eth_getUserOperationReceipt" | "eth_getUserOperationReceipt"
| "eth_supportedEntryPoints") => match self.bundler_4337_rpcs.as_ref() { | "eth_supportedEntryPoints") => match self.bundler_4337_rpcs.as_ref() {
Some(bundler_4337_rpcs) => { Some(bundler_4337_rpcs) => {
// TODO: timeout
bundler_4337_rpcs bundler_4337_rpcs
.try_proxy_connection( .try_proxy_connection(
authorization, authorization,
@ -1367,6 +1358,7 @@ impl Web3ProxyApp {
JsonRpcResponseData::from(json!(Address::zero())) JsonRpcResponseData::from(json!(Address::zero()))
} }
"eth_estimateGas" => { "eth_estimateGas" => {
// TODO: timeout
let response_data = self let response_data = self
.balanced_rpcs .balanced_rpcs
.try_proxy_connection( .try_proxy_connection(
@ -1403,6 +1395,7 @@ impl Web3ProxyApp {
} }
"eth_getTransactionReceipt" | "eth_getTransactionByHash" => { "eth_getTransactionReceipt" | "eth_getTransactionByHash" => {
// try to get the transaction without specifying a min_block_height // try to get the transaction without specifying a min_block_height
// TODO: timeout
let mut response_data = self let mut response_data = self
.balanced_rpcs .balanced_rpcs
.try_proxy_connection( .try_proxy_connection(
@ -1446,12 +1439,7 @@ impl Web3ProxyApp {
// TODO: error if the chain_id is incorrect // TODO: error if the chain_id is incorrect
// TODO: check the cache to see if we have sent this transaction recently // TODO: timeout
// TODO: if so, use a cached response.
// TODO: if not,
// TODO: - cache successes for up to 1 minute
// TODO: - cache failures for 1 block (we see transactions skipped because out of funds. but that can change block to block)
let mut response_data = self let mut response_data = self
.try_send_protected( .try_send_protected(
authorization, authorization,
@ -1581,6 +1569,7 @@ impl Web3ProxyApp {
, ,
"web3_sha3" => { "web3_sha3" => {
// returns Keccak-256 (not the standardized SHA3-256) of the given data. // returns Keccak-256 (not the standardized SHA3-256) of the given data.
// TODO: timeout
match &request.params { match &request.params {
Some(serde_json::Value::Array(params)) => { Some(serde_json::Value::Array(params)) => {
// TODO: make a struct and use serde conversion to clean this up // TODO: make a struct and use serde conversion to clean this up
@ -1740,6 +1729,9 @@ impl Web3ProxyApp {
let authorization = authorization.clone(); let authorization = authorization.clone();
// TODO: different timeouts for different user tiers
let duration = Duration::from_secs(240);
if let Some(cache_key) = cache_key { if let Some(cache_key) = cache_key {
let from_block_num = cache_key.from_block.as_ref().map(|x| x.number.unwrap()); let from_block_num = cache_key.from_block.as_ref().map(|x| x.number.unwrap());
let to_block_num = cache_key.to_block.as_ref().map(|x| x.number.unwrap()); let to_block_num = cache_key.to_block.as_ref().map(|x| x.number.unwrap());
@ -1750,7 +1742,9 @@ impl Web3ProxyApp {
{ {
Ok(x) => x, Ok(x) => x,
Err(x) => { Err(x) => {
let response_data = self.balanced_rpcs let response_data = timeout(
duration,
self.balanced_rpcs
.try_proxy_connection( .try_proxy_connection(
&authorization, &authorization,
request, request,
@ -1758,7 +1752,8 @@ impl Web3ProxyApp {
from_block_num.as_ref(), from_block_num.as_ref(),
to_block_num.as_ref(), to_block_num.as_ref(),
) )
.await?; )
.await??;
// TODO: convert the Box<RawValue> to an Arc<RawValue> // TODO: convert the Box<RawValue> to an Arc<RawValue>
x.insert(response_data.clone()); x.insert(response_data.clone());
@ -1767,6 +1762,8 @@ impl Web3ProxyApp {
} }
} }
} else { } else {
timeout(
duration,
self.balanced_rpcs self.balanced_rpcs
.try_proxy_connection( .try_proxy_connection(
&authorization, &authorization,
@ -1775,7 +1772,8 @@ impl Web3ProxyApp {
None, None,
None, None,
) )
.await? )
.await??
} }
} }
}; };