provider.broken_provider

Documentation for eth_defi.provider.broken_provider Python module.

Fixes and workaronds for various JSON-RPC service providers.

  • Mainly to deal with unstable blockchain chain tip

See also

Functions

get_almost_latest_block_number(web3)

Get the latest block number with workarounds for low quality JSON-RPC service providers.

get_block_tip_latency(web3)

What block tip latency we should use for this JSON-RPC provider

get_default_block_tip_latency(web3)

Workaround for Ankr and other node providers that do not handle the chain tip properly.

get_fork_safe_latest_block(web3)

Get the latest block identifier that is safe for Anvil mainnet forks.

get_safe_cached_latest_block_number(web3, ...)

Get almost "latest" block to work around broken JSON-RPC providers.

set_block_tip_latency(web3, block_count)

Override the default block tip latency settings.

verify_archive_node(rpc_url, chain_name)

Verify RPC providers and filter out broken ones.
get_default_block_tip_latency(web3)

Workaround for Ankr and other node providers that do not handle the chain tip properly.

Likely due to how requests are broken a block number available in one call disappears in the following. Thus, with broken JSON-RPC service providers we cannot trust web3.eth.block_number to work and we need to subtract this number from it.

See get_block_tip_latency()

Returns

Number of blocks we need to subtract from the latest block

Parameters

web3 (web3.main.Web3) –

Return type

int

get_block_tip_latency(web3)

What block tip latency we should use for this JSON-RPC provider

  • Defaults to zero

  • If using eth_defi.provider.fallback.FallbackProvider we use 4 blocks latency as multiple providers are unlikely to agree on a chain tip (blocks have not propagated yet).

  • We have some built-in rules to work out specific providers

  • You can override this by setting the latency sa web3.block_tip_latency attribute

See the source code of get_default_block_tip_latency() for other rules.

Parameters

web3 (web3.main.Web3) –

Return type

int

set_block_tip_latency(web3, block_count)

Override the default block tip latency settings.

Useful for unit testing, because unit testing assumes stuff has happened in the latest block you want to read.

See get_block_tip_latency().

Parameters

  • web3 (web3.main.Web3) –

  • block_count (int) –

get_fork_safe_latest_block(web3)

Get the latest block identifier that is safe for Anvil mainnet forks.

  • For Anvil mainnet forks, returns web3.eth.block_number (an integer) instead of "latest" to prevent the upstream RPC from resolving "latest" to the actual chain tip (which may be beyond the fork block or the upstream’s available window).

  • For non-fork Anvil (test backend), returns "latest" as there is no upstream RPC.

  • For non-Anvil providers, returns "latest".

This is needed because web3.py v7 forwards "latest" to the upstream RPC when Anvil needs to fetch state not cached at fork time. The upstream then resolves "latest" to its own chain tip, causing BlockOutOfRangeError if the upstream has a limited block window.

See get_safe_cached_latest_block_number() which uses this function.

Parameters

web3 (web3.main.Web3) –

Return type

Union[Literal[‘latest’, ‘earliest’, ‘pending’, ‘safe’, ‘finalized’], eth_typing.evm.BlockNumber, eth_typing.evm.Hash32, eth_typing.encoding.HexStr, int]

get_almost_latest_block_number(web3)

Get the latest block number with workarounds for low quality JSON-RPC service providers.

Use this method instead of web3.eth.block_number.

Because low quality providers may lose the block of this block number on the subsequent API calls, we add some number of delay or confirmations to the chain tip, specified by get_block_tip_latency().

Providers with known issues

  • LlamaNodes

  • Ankr

Example:

from eth_defi.provider.broken_provider import get_almost_latest_block_number

# We cannot query the chain head on Ankr or LlamaNodes,
# so get the almost head
if not block_number:
    block_number = get_almost_latest_block_number(web3)

timestamp = fetch_block_timestamp(web3, block_number)

token = fetch_erc20_details(web3, asset.address)
amount = token.fetch_balance_of(address, block_identifier=block_number)
Parameters

web3 (web3.main.Web3) –

Return type

int

get_safe_cached_latest_block_number(web3, chain_id, blocks=1000, cache_duration=3600)

Get almost “latest” block to work around broken JSON-RPC providers.

  • Not for high frequency usage, as it caches the block for delay seconds

  • No RPC call are made to

  • Disabled in Anvil configs

Work around the error on Monad/Arbitrum/dRPC/shitty RPCs:

{'message': 'upstream does not have the requested block yet', 'code': -32603}

Their internal routing is likely broken and when calling eth_call with latest the request fails for no reason.

Parameters

  • chain_id (int) – Chain id to use as part of the cache key

  • blocks – Number of blocks to subtract from the latest block

  • cache_duration (int) – Number of seconds to cache the result

  • web3 (web3.main.Web3) –

Returns

Latest block number minus blocks.

May return “latest” for special configs like unit tests.

Return type

Union[Literal[‘latest’, ‘earliest’, ‘pending’, ‘safe’, ‘finalized’], eth_typing.evm.BlockNumber, eth_typing.evm.Hash32, eth_typing.encoding.HexStr, int]

verify_archive_node(rpc_url, chain_name)

Verify RPC providers and filter out broken ones.

Parses the space-separated multi-RPC configuration line and tests each endpoint individually. Checks that each provider can serve both block 1 and the latest block. Broken providers are logged at ERROR level and filtered out; the scan continues with working providers only.

Parameters

  • rpc_url (str) – RPC URL configuration line (may contain space-separated fallbacks, mev+ prefixed endpoints are skipped)

  • chain_name (str) – Chain name for logging

Returns

Tuple of (filtered RPC URL with only working providers, latest block number)

Raises

RuntimeError – If all providers fail verification

Return type

tuple[str, int]