diff --git a/config b/config index 6e34c5b6..a7048273 160000 --- a/config +++ b/config @@ -1 +1 @@ -Subproject commit 6e34c5b68fa3fba7cad3b140f8676dcbdab687c5 +Subproject commit a70482731c55f8c2079da0df66d084c2468bf3ad diff --git a/snapshotter/modules/computes b/snapshotter/modules/computes index b4295c41..e890277c 160000 --- a/snapshotter/modules/computes +++ b/snapshotter/modules/computes @@ -1 +1 @@ -Subproject commit b4295c41b81ca0f8ba16df7200620c72dc5afffb +Subproject commit e890277c451f4724800b28c010a24af29c0de53f diff --git a/snapshotter/utils/rpc.py b/snapshotter/utils/rpc.py index 801548f6..03eff492 100644 --- a/snapshotter/utils/rpc.py +++ b/snapshotter/utils/rpc.py @@ -344,7 +344,7 @@ async def f(node_idx): return current_block return await f(node_idx=0) - async def _async_web3_call(self, contract_function, redis_conn, from_address=None): + async def _async_web3_call(self, contract_function, redis_conn, from_address=None, block=None, overrides=None): """ Executes a web3 call asynchronously. @@ -365,43 +365,42 @@ async def _async_web3_call(self, contract_function, redis_conn, from_address=Non ) async def f(node_idx): try: - node = self._nodes[node_idx] - rpc_url = node.get('rpc_url') + rpc_url = node.get("rpc_url") await check_rpc_rate_limit( - parsed_limits=node.get('rate_limit', []), - app_id=rpc_url.split('/')[-1], + parsed_limits=node.get("rate_limit", []), + app_id=rpc_url.split("/")[-1], redis_conn=redis_conn, request_payload=contract_function.fn_name, error_msg={ - 'msg': 'exhausted_api_key_rate_limit inside web3_call', + "msg": "exhausted_api_key_rate_limit inside web3_call", }, logger=self._logger, rate_limit_lua_script_shas=self._rate_limit_lua_script_shas, limit_incr_by=1, ) - - params: TxParams = {'gas': Wei(0), 'gasPrice': Wei(0)} + # NOTE is there a reason this is set to 0? + params: TxParams = {"gas": Wei(0), "gasPrice": Wei(0)} if not contract_function.address: raise ValueError( - f'Missing address for batch_call in `{contract_function.fn_name}`', + f"Missing address for batch_call in `{[contract_function.fn_name]}`", ) output_type = [ - output['type'] for output in contract_function.abi['outputs'] + output["type"] for output in contract_function.abi["outputs"] ] payload = { - 'to': contract_function.address, - 'data': contract_function.build_transaction(params)['data'], - 'output_type': output_type, - 'fn_name': contract_function.fn_name, # For debugging purposes + "to": contract_function.address, + "data": contract_function.build_transaction(params)["data"], + "output_type": output_type, + "fn_name": contract_function.fn_name, # For debugging purposes } cur_time = time.time() redis_cache_data = payload.copy() - redis_cache_data['time'] = cur_time + redis_cache_data["time"] = cur_time await asyncio.gather( redis_conn.zadd( name=rpc_web3_calls, @@ -417,16 +416,26 @@ async def f(node_idx): ) if from_address: - payload['from'] = from_address - - data = await node['web3_client_async'].eth.call(payload) + payload["from"] = from_address - decoded_data = node['web3_client_async'].codec.decode_abi( - output_type, HexBytes(data), + data = await node["web3_client_async"].eth.call( + payload, block_identifier=block, state_override=overrides + ) + + # if we're doing a state override call, at time of writing it means grabbing tick data + # more efficient to use eth_abi to decode rather than web3 codec + if overrides is not None: + return data + + decoded_data = node["web3_client_async"].codec.decode( + output_type, + HexBytes(data), ) normalized_data = map_abi_data( - BASE_RETURN_NORMALIZERS, output_type, decoded_data, + BASE_RETURN_NORMALIZERS, + output_type, + decoded_data, ) if len(normalized_data) == 1: @@ -438,12 +447,11 @@ async def f(node_idx): request=[contract_function.fn_name], response=None, underlying_exception=e, - extra_info={'msg': str(e)}, + extra_info={"msg": str(e)}, ) - self._logger.opt(exception=settings.logs.trace_enabled).error( - ( - 'Error while making web3 batch call' - ), + + self._logger.opt(lazy=True).trace( + ("Error while making web3 batch call"), err=lambda: str(exc), ) raise exc @@ -577,7 +585,7 @@ async def get_current_block(self, redis_conn, node_idx=0): ) return current_block - async def web3_call(self, tasks, redis_conn, from_address=None): + async def web3_call(self, tasks, redis_conn, from_address=None, block=None, overrides=None): """ Calls the given tasks asynchronously using web3 and returns the response. @@ -595,8 +603,13 @@ async def web3_call(self, tasks, redis_conn, from_address=None): try: web3_tasks = [ self._async_web3_call( - contract_function=task, redis_conn=redis_conn, from_address=from_address, - ) for task in tasks + contract_function=task, + redis_conn=redis_conn, + from_address=from_address, + block=block, + overrides=overrides + ) + for task in tasks ] response = await asyncio.gather(*web3_tasks) return response