From ca3a30c5f0cb6e86a44aa10e39f246de11a352a1 Mon Sep 17 00:00:00 2001 From: Zbynek Masler Date: Thu, 10 Sep 2026 11:04:38 +0200 Subject: [PATCH 1/4] fix(solana): slow fetches and retry Helius 429 responses --- blockapi/test/v2/api/test_solana.py | 222 ++++++++++++++++++++++++++-- blockapi/v2/api/solana.py | 54 ++++++- 2 files changed, 259 insertions(+), 17 deletions(-) diff --git a/blockapi/test/v2/api/test_solana.py b/blockapi/test/v2/api/test_solana.py index 8202dc7..632986a 100644 --- a/blockapi/test/v2/api/test_solana.py +++ b/blockapi/test/v2/api/test_solana.py @@ -1,4 +1,5 @@ import json +import logging from decimal import Decimal from unittest.mock import call, patch @@ -6,8 +7,9 @@ from requests_mock import ANY, Mocker from blockapi.test.v2.api.conftest import read_file +from blockapi.test.v2.api.fake_sleep_provider import FakeSleepProvider from blockapi.v2.api import SolanaApi, SolscanApi -from blockapi.v2.base import ApiException +from blockapi.v2.base import ApiException, InvalidAddressException from blockapi.v2.models import ( AssetType, BalanceItem, @@ -77,10 +79,15 @@ def test_use_base_url(): @pytest.mark.parametrize( - ('rpc_url', 'uses_v2_staking'), + ('rpc_url', 'uses_v2_staking', 'throttled_request_index'), [ - ('https://mainnet.helius-rpc.com/', True), - ('https://proxy/solana/', False), + ('https://mainnet.helius-rpc.com/', True, None), + ('https://proxy/solana/', False, None), + ('https://mainnet.helius-rpc.com/', True, 0), + ('https://mainnet.helius-rpc.com/', True, 1), + ('https://mainnet.helius-rpc.com/', True, 2), + ('https://mainnet.helius-rpc.com/', True, 3), + ('https://mainnet.helius-rpc.com/', True, 4), ], ) def test_get_balance_supports_helius_and_legacy_staking_responses( @@ -90,6 +97,8 @@ def test_get_balance_supports_helius_and_legacy_staking_responses( staked_solana_response, rpc_url, uses_v2_staking, + throttled_request_index, + caplog, ): test_addr = '5PjMxaijeVVQtuEzxK2NxyJeWwUbpTsi2uXuZ653WoHu' empty_token_accounts = '{"jsonrpc":"2.0","result":{"context":{"apiVersion":"1.17.34","slot":268207149},"value":[]},"id":1}' @@ -107,16 +116,45 @@ def test_get_balance_supports_helius_and_legacy_staking_responses( ] ) - def get_text(*args, **kwargs): - assert args[0].url == rpc_url - data = next(iterator) - return data + methods = [] + sleep_provider = FakeSleepProvider() - with Mocker() as m: + def get_text(request, context): + assert request.url == rpc_url + method = request.json()['method'] + methods.append(method) + if len(methods) - 1 == throttled_request_index: + context.status_code = 429 + return 'Too Many Requests' + return next(iterator) + + with Mocker() as m, patch('time.sleep') as sleep: m.post(ANY, text=get_text) - api = SolanaApi(base_url=rpc_url) + api = SolanaApi(base_url=rpc_url, sleep_provider=sleep_provider) balances = api.get_balance(test_addr) + expected_methods = [ + 'getBalance', + 'getTokenAccountsByOwner', + 'getTokenAccountsByOwner', + 'getAssetBatch', + 'getProgramAccountsV2' if uses_v2_staking else 'getProgramAccounts', + ] + if throttled_request_index is not None: + retry_index = throttled_request_index + expected_methods.insert(retry_index, expected_methods[retry_index]) + assert ( + m.request_history[retry_index].json() + == m.request_history[retry_index + 1].json() + ) + sleep.assert_called_once() + assert 1 <= sleep.call_args.args[0] <= 1.25 + else: + sleep.assert_not_called() + assert methods == expected_methods + assert sleep_provider.calls == [] + assert not any(record.levelno >= logging.ERROR for record in caplog.records) + staking_balances = { balance.asset_type: balance.balance_raw for balance in balances @@ -128,6 +166,170 @@ def get_text(*args, **kwargs): } +def test_get_balance_fails_after_three_throttled_token_attempts(requests_mock): + rpc_url = 'https://mainnet.helius-rpc.com/' + requests_mock.post( + rpc_url, + [ + {'json': {'result': {'value': 1}}}, + {'status_code': 429, 'text': 'Too Many Requests'}, + ], + ) + sleep_provider = FakeSleepProvider() + api = SolanaApi(base_url=rpc_url, sleep_provider=sleep_provider) + + with patch('time.sleep') as sleep, pytest.raises(ApiException, match='429'): + api.get_balance('address') + + assert [r.json()['method'] for r in requests_mock.request_history] == [ + 'getBalance', + 'getTokenAccountsByOwner', + 'getTokenAccountsByOwner', + 'getTokenAccountsByOwner', + ] + assert sleep_provider.calls == [] + assert len(sleep.call_args_list) == 2 + assert 1 <= sleep.call_args_list[0].args[0] <= 1.25 + assert 2 <= sleep.call_args_list[1].args[0] <= 2.25 + + +@pytest.mark.parametrize( + ('retry_after', 'min_delay'), + [ + (None, 1), + ('2', 2), + ('5', 5), + ('0', 1), + ('-1', 1), + ('invalid', 1), + ('Thu, 10 Sep 2026 07:00:03 GMT', 3), + ('Thu, 10 Sep 2026 06:59:59 GMT', 1), + ], +) +def test_helius_retry_after_without_sleep_provider( + requests_mock, retry_after, min_delay +): + rpc_url = 'https://mainnet.helius-rpc.com/?api-key=test-key' + headers = {'Retry-After': retry_after} if retry_after is not None else {} + requests_mock.post( + rpc_url, + [ + {'status_code': 429, 'text': 'Too Many Requests', 'headers': headers}, + {'json': {'result': {'value': 1}}}, + ], + ) + api = SolanaApi(base_url=rpc_url) + + with patch('time.sleep') as sleep, patch('time.time', return_value=1789023600): + result = api._request('getBalance', ['address']) + + assert result == {'result': {'value': 1}} + assert requests_mock.call_count == 2 + sleep.assert_called_once() + assert min_delay <= sleep.call_args.args[0] <= min(min_delay + 0.25, 5) + + +@pytest.mark.parametrize('retry_after', ['6', '60', 'Thu, 10 Sep 2026 07:01:00 GMT']) +def test_helius_long_retry_after_fails_without_early_retry(requests_mock, retry_after): + rpc_url = 'https://mainnet.helius-rpc.com/' + requests_mock.post( + rpc_url, + status_code=429, + text='Too Many Requests', + headers={'Retry-After': retry_after}, + ) + api = SolanaApi(base_url=rpc_url) + + with patch('time.sleep') as sleep, patch('time.time', return_value=1789023600): + with pytest.raises(ApiException, match='429'): + api.get_balance('address') + + assert requests_mock.call_count == 1 + sleep.assert_not_called() + + +@pytest.mark.parametrize('status_code', [400, 401, 403, 500, 503]) +def test_helius_does_not_retry_other_http_errors(requests_mock, status_code): + rpc_url = 'https://mainnet.helius-rpc.com/' + requests_mock.post(rpc_url, status_code=status_code, text='RPC unavailable') + api = SolanaApi(base_url=rpc_url) + + with patch('time.sleep') as sleep: + with pytest.raises(ApiException, match=str(status_code)): + api.get_balance('address') + + assert requests_mock.call_count == 1 + sleep.assert_not_called() + + +def test_helius_keeps_invalid_address_error(requests_mock): + rpc_url = 'https://mainnet.helius-rpc.com/' + requests_mock.post( + rpc_url, + json={'error': {'code': -32602, 'message': 'Invalid param: WrongSize'}}, + ) + api = SolanaApi(base_url=rpc_url) + + with patch('time.sleep') as sleep, pytest.raises(InvalidAddressException): + api.get_balance('invalid') + + assert requests_mock.call_count == 1 + sleep.assert_not_called() + + +@pytest.mark.parametrize( + 'rpc_url', ['https://proxy/solana/', 'https://helius-rpc.com.example.org/'] +) +def test_non_helius_throttling_keeps_existing_behavior(requests_mock, rpc_url): + requests_mock.post(rpc_url, status_code=429, text='Too Many Requests') + sleep_provider = FakeSleepProvider() + api = SolanaApi(base_url=rpc_url, sleep_provider=sleep_provider) + + with patch('time.sleep') as sleep, pytest.raises(ApiException, match='429'): + api.get_balance('address') + + assert requests_mock.call_count == 1 + assert sleep_provider.calls == [] + sleep.assert_not_called() + + +def test_helius_retries_staking_page_without_duplicate_accounts(requests_mock): + rpc_url = 'https://mainnet.helius-rpc.com/' + requests_mock.post( + rpc_url, + [ + { + 'json': { + 'result': { + 'accounts': [{'pubkey': 'first'}], + 'paginationKey': 'next', + } + } + }, + {'status_code': 429, 'text': 'Too Many Requests'}, + { + 'json': { + 'result': { + 'accounts': [{'pubkey': 'second'}], + 'paginationKey': None, + } + } + }, + ], + ) + api = SolanaApi(base_url=rpc_url) + + with patch('time.sleep'): + result = api._fetch_staked_sol('address') + + assert result['result'] == [{'pubkey': 'first'}, {'pubkey': 'second'}] + assert requests_mock.call_count == 3 + assert 'paginationKey' not in requests_mock.request_history[0].json()['params'][1] + retry_body = requests_mock.request_history[1].json() + assert retry_body['params'][1]['paginationKey'] == 'next' + assert requests_mock.request_history[2].json() == retry_body + + def test_build_coin_from_das_asset(): api = SolanaApi() asset = { diff --git a/blockapi/v2/api/solana.py b/blockapi/v2/api/solana.py index 29a141f..ab3f8b0 100644 --- a/blockapi/v2/api/solana.py +++ b/blockapi/v2/api/solana.py @@ -1,5 +1,8 @@ import json import logging +import random +import time +from email.utils import parsedate_to_datetime from typing import Optional, Union from urllib.parse import urlparse @@ -67,7 +70,7 @@ class SolanaApi(CustomizableBlockchainApi, BalanceMixin): api_options = ApiOptions( blockchain=Blockchain.SOLANA, base_url='https://api.mainnet-beta.solana.com/', - rate_limit=1, + rate_limit=2, start_offset=0, max_items_per_page=1000, page_offset_step=1, @@ -82,6 +85,8 @@ class SolanaApi(CustomizableBlockchainApi, BalanceMixin): STAKE_AUTHORITY_OFFSET = 44 HELIUS_RPC_DOMAIN = 'helius-rpc.com' HELIUS_PROGRAM_ACCOUNTS_PAGE_SIZE = 5000 + HELIUS_MAX_ATTEMPTS = 3 + HELIUS_MAX_RETRY_DELAY = 5 DAS_BATCH_SIZE = 1000 _JSONRPC_INVALID_PARAMS = -32602 @@ -331,11 +336,7 @@ def _fetch_staked_sol(self, address: str) -> dict: 'commitment': 'finalized', } - hostname = urlparse(self.base_url).hostname or '' - is_helius_rpc = hostname == self.HELIUS_RPC_DOMAIN or hostname.endswith( - f'.{self.HELIUS_RPC_DOMAIN}' - ) - if not is_helius_rpc: + if not self._is_helius_rpc(): return self._request( method='getProgramAccounts', params=[self.STAKE_PROGRAM_ID, config], @@ -430,7 +431,46 @@ def _request(self, method: str, params: Union[list, dict]) -> dict: 'params': params, } ) - return self.post(body=body, headers={'Content-Type': 'application/json'}) + headers = {'Content-Type': 'application/json'} + if not self._is_helius_rpc(): + return self.post(body=body, headers=headers) + + for attempt in range(self.HELIUS_MAX_ATTEMPTS): + response = self._session.post(self.base_url, data=body, headers=headers) + if response.status_code != 429 or attempt == self.HELIUS_MAX_ATTEMPTS - 1: + break + + delay = self._helius_retry_delay(response, attempt) + # Leave long cooldowns to the caller's next fetch instead of blocking it. + if delay > self.HELIUS_MAX_RETRY_DELAY: + break + delay = min(delay + random.uniform(0, 0.25), self.HELIUS_MAX_RETRY_DELAY) + logger.warning(f'Helius {method} rate limited; retrying in {delay:.2f}s') + response.close() + time.sleep(delay) + + return self._check_and_get_from_response(response) + + def _is_helius_rpc(self) -> bool: + hostname = urlparse(self.base_url).hostname or '' + return hostname == self.HELIUS_RPC_DOMAIN or hostname.endswith( + f'.{self.HELIUS_RPC_DOMAIN}' + ) + + @staticmethod + def _helius_retry_delay(response: Response, attempt: int) -> float: + delay = 2**attempt + retry_after = response.headers.get('Retry-After') + if retry_after: + try: + return max(delay, int(retry_after)) + except ValueError: + try: + retry_at = parsedate_to_datetime(retry_after).timestamp() + return max(delay, retry_at - time.time()) + except (TypeError, ValueError, OverflowError): + pass + return delay def _opt_raise_on_other_error(self, response: Response) -> None: """Raise ApiException or InvalidAddressException on RPC errors.""" From b3b3d8acd0418f25d6da25d14c30b82bcc1a1862 Mon Sep 17 00:00:00 2001 From: Zbynek Masler Date: Thu, 10 Sep 2026 11:57:14 +0200 Subject: [PATCH 2/4] test(solana): simplify mocks and cover final retry success --- blockapi/test/v2/api/test_solana.py | 83 +++++++++++++++-------------- 1 file changed, 42 insertions(+), 41 deletions(-) diff --git a/blockapi/test/v2/api/test_solana.py b/blockapi/test/v2/api/test_solana.py index 632986a..f48caaf 100644 --- a/blockapi/test/v2/api/test_solana.py +++ b/blockapi/test/v2/api/test_solana.py @@ -4,7 +4,6 @@ from unittest.mock import call, patch import pytest -from requests_mock import ANY, Mocker from blockapi.test.v2.api.conftest import read_file from blockapi.test.v2.api.fake_sleep_provider import FakeSleepProvider @@ -91,6 +90,7 @@ def test_use_base_url(): ], ) def test_get_balance_supports_helius_and_legacy_staking_responses( + requests_mock, sol_balance_response, token_accounts_response, das_asset_batch_response, @@ -106,30 +106,22 @@ def test_get_balance_supports_helius_and_legacy_staking_responses( if not uses_v2_staking: staking_response['result'] = staking_response['result']['accounts'] - iterator = iter( - [ - sol_balance_response, - token_accounts_response, - empty_token_accounts, - das_asset_batch_response, - json.dumps(staking_response), - ] - ) - - methods = [] + responses = [ + {'text': sol_balance_response}, + {'text': token_accounts_response}, + {'text': empty_token_accounts}, + {'text': das_asset_batch_response}, + {'json': staking_response}, + ] + if throttled_request_index is not None: + responses.insert( + throttled_request_index, + {'status_code': 429, 'text': 'Too Many Requests'}, + ) + requests_mock.post(rpc_url, responses) sleep_provider = FakeSleepProvider() - def get_text(request, context): - assert request.url == rpc_url - method = request.json()['method'] - methods.append(method) - if len(methods) - 1 == throttled_request_index: - context.status_code = 429 - return 'Too Many Requests' - return next(iterator) - - with Mocker() as m, patch('time.sleep') as sleep: - m.post(ANY, text=get_text) + with patch('time.sleep') as sleep: api = SolanaApi(base_url=rpc_url, sleep_provider=sleep_provider) balances = api.get_balance(test_addr) @@ -144,14 +136,16 @@ def get_text(request, context): retry_index = throttled_request_index expected_methods.insert(retry_index, expected_methods[retry_index]) assert ( - m.request_history[retry_index].json() - == m.request_history[retry_index + 1].json() + requests_mock.request_history[retry_index].json() + == requests_mock.request_history[retry_index + 1].json() ) sleep.assert_called_once() assert 1 <= sleep.call_args.args[0] <= 1.25 else: sleep.assert_not_called() - assert methods == expected_methods + assert [ + r.json()['method'] for r in requests_mock.request_history + ] == expected_methods assert sleep_provider.calls == [] assert not any(record.levelno >= logging.ERROR for record in caplog.records) @@ -194,20 +188,21 @@ def test_get_balance_fails_after_three_throttled_token_attempts(requests_mock): @pytest.mark.parametrize( - ('retry_after', 'min_delay'), + ('retry_after', 'min_delays'), [ - (None, 1), - ('2', 2), - ('5', 5), - ('0', 1), - ('-1', 1), - ('invalid', 1), - ('Thu, 10 Sep 2026 07:00:03 GMT', 3), - ('Thu, 10 Sep 2026 06:59:59 GMT', 1), + (None, [1]), + pytest.param(None, [1, 2], id='success-on-third-attempt'), + ('2', [2]), + ('5', [5]), + ('0', [1]), + ('-1', [1]), + ('invalid', [1]), + ('Thu, 10 Sep 2026 07:00:03 GMT', [3]), + ('Thu, 10 Sep 2026 06:59:59 GMT', [1]), ], ) def test_helius_retry_after_without_sleep_provider( - requests_mock, retry_after, min_delay + requests_mock, retry_after, min_delays ): rpc_url = 'https://mainnet.helius-rpc.com/?api-key=test-key' headers = {'Retry-After': retry_after} if retry_after is not None else {} @@ -215,8 +210,9 @@ def test_helius_retry_after_without_sleep_provider( rpc_url, [ {'status_code': 429, 'text': 'Too Many Requests', 'headers': headers}, - {'json': {'result': {'value': 1}}}, - ], + ] + * len(min_delays) + + [{'json': {'result': {'value': 1}}}], ) api = SolanaApi(base_url=rpc_url) @@ -224,9 +220,14 @@ def test_helius_retry_after_without_sleep_provider( result = api._request('getBalance', ['address']) assert result == {'result': {'value': 1}} - assert requests_mock.call_count == 2 - sleep.assert_called_once() - assert min_delay <= sleep.call_args.args[0] <= min(min_delay + 0.25, 5) + assert requests_mock.call_count == len(min_delays) + 1 + assert sleep.call_count == len(min_delays) + for sleep_call, min_delay in zip(sleep.call_args_list, min_delays): + assert min_delay <= sleep_call.args[0] <= min(min_delay + 0.25, 5) + assert all( + request.json() == requests_mock.request_history[0].json() + for request in requests_mock.request_history[1:] + ) @pytest.mark.parametrize('retry_after', ['6', '60', 'Thu, 10 Sep 2026 07:01:00 GMT']) From 1a409569866ac69a9e8f8253ffb4027f33401847 Mon Sep 17 00:00:00 2001 From: Zbynek Masler Date: Thu, 10 Sep 2026 12:25:43 +0200 Subject: [PATCH 3/4] fix(solana): simplify Helius mitigation to a two-second fetch interval --- blockapi/test/v2/api/test_solana.py | 249 +++------------------------- blockapi/v2/api/solana.py | 52 +----- 2 files changed, 29 insertions(+), 272 deletions(-) diff --git a/blockapi/test/v2/api/test_solana.py b/blockapi/test/v2/api/test_solana.py index f48caaf..8202dc7 100644 --- a/blockapi/test/v2/api/test_solana.py +++ b/blockapi/test/v2/api/test_solana.py @@ -1,14 +1,13 @@ import json -import logging from decimal import Decimal from unittest.mock import call, patch import pytest +from requests_mock import ANY, Mocker from blockapi.test.v2.api.conftest import read_file -from blockapi.test.v2.api.fake_sleep_provider import FakeSleepProvider from blockapi.v2.api import SolanaApi, SolscanApi -from blockapi.v2.base import ApiException, InvalidAddressException +from blockapi.v2.base import ApiException from blockapi.v2.models import ( AssetType, BalanceItem, @@ -78,27 +77,19 @@ def test_use_base_url(): @pytest.mark.parametrize( - ('rpc_url', 'uses_v2_staking', 'throttled_request_index'), + ('rpc_url', 'uses_v2_staking'), [ - ('https://mainnet.helius-rpc.com/', True, None), - ('https://proxy/solana/', False, None), - ('https://mainnet.helius-rpc.com/', True, 0), - ('https://mainnet.helius-rpc.com/', True, 1), - ('https://mainnet.helius-rpc.com/', True, 2), - ('https://mainnet.helius-rpc.com/', True, 3), - ('https://mainnet.helius-rpc.com/', True, 4), + ('https://mainnet.helius-rpc.com/', True), + ('https://proxy/solana/', False), ], ) def test_get_balance_supports_helius_and_legacy_staking_responses( - requests_mock, sol_balance_response, token_accounts_response, das_asset_batch_response, staked_solana_response, rpc_url, uses_v2_staking, - throttled_request_index, - caplog, ): test_addr = '5PjMxaijeVVQtuEzxK2NxyJeWwUbpTsi2uXuZ653WoHu' empty_token_accounts = '{"jsonrpc":"2.0","result":{"context":{"apiVersion":"1.17.34","slot":268207149},"value":[]},"id":1}' @@ -106,48 +97,25 @@ def test_get_balance_supports_helius_and_legacy_staking_responses( if not uses_v2_staking: staking_response['result'] = staking_response['result']['accounts'] - responses = [ - {'text': sol_balance_response}, - {'text': token_accounts_response}, - {'text': empty_token_accounts}, - {'text': das_asset_batch_response}, - {'json': staking_response}, - ] - if throttled_request_index is not None: - responses.insert( - throttled_request_index, - {'status_code': 429, 'text': 'Too Many Requests'}, - ) - requests_mock.post(rpc_url, responses) - sleep_provider = FakeSleepProvider() - - with patch('time.sleep') as sleep: - api = SolanaApi(base_url=rpc_url, sleep_provider=sleep_provider) - balances = api.get_balance(test_addr) + iterator = iter( + [ + sol_balance_response, + token_accounts_response, + empty_token_accounts, + das_asset_batch_response, + json.dumps(staking_response), + ] + ) - expected_methods = [ - 'getBalance', - 'getTokenAccountsByOwner', - 'getTokenAccountsByOwner', - 'getAssetBatch', - 'getProgramAccountsV2' if uses_v2_staking else 'getProgramAccounts', - ] - if throttled_request_index is not None: - retry_index = throttled_request_index - expected_methods.insert(retry_index, expected_methods[retry_index]) - assert ( - requests_mock.request_history[retry_index].json() - == requests_mock.request_history[retry_index + 1].json() - ) - sleep.assert_called_once() - assert 1 <= sleep.call_args.args[0] <= 1.25 - else: - sleep.assert_not_called() - assert [ - r.json()['method'] for r in requests_mock.request_history - ] == expected_methods - assert sleep_provider.calls == [] - assert not any(record.levelno >= logging.ERROR for record in caplog.records) + def get_text(*args, **kwargs): + assert args[0].url == rpc_url + data = next(iterator) + return data + + with Mocker() as m: + m.post(ANY, text=get_text) + api = SolanaApi(base_url=rpc_url) + balances = api.get_balance(test_addr) staking_balances = { balance.asset_type: balance.balance_raw @@ -160,177 +128,6 @@ def test_get_balance_supports_helius_and_legacy_staking_responses( } -def test_get_balance_fails_after_three_throttled_token_attempts(requests_mock): - rpc_url = 'https://mainnet.helius-rpc.com/' - requests_mock.post( - rpc_url, - [ - {'json': {'result': {'value': 1}}}, - {'status_code': 429, 'text': 'Too Many Requests'}, - ], - ) - sleep_provider = FakeSleepProvider() - api = SolanaApi(base_url=rpc_url, sleep_provider=sleep_provider) - - with patch('time.sleep') as sleep, pytest.raises(ApiException, match='429'): - api.get_balance('address') - - assert [r.json()['method'] for r in requests_mock.request_history] == [ - 'getBalance', - 'getTokenAccountsByOwner', - 'getTokenAccountsByOwner', - 'getTokenAccountsByOwner', - ] - assert sleep_provider.calls == [] - assert len(sleep.call_args_list) == 2 - assert 1 <= sleep.call_args_list[0].args[0] <= 1.25 - assert 2 <= sleep.call_args_list[1].args[0] <= 2.25 - - -@pytest.mark.parametrize( - ('retry_after', 'min_delays'), - [ - (None, [1]), - pytest.param(None, [1, 2], id='success-on-third-attempt'), - ('2', [2]), - ('5', [5]), - ('0', [1]), - ('-1', [1]), - ('invalid', [1]), - ('Thu, 10 Sep 2026 07:00:03 GMT', [3]), - ('Thu, 10 Sep 2026 06:59:59 GMT', [1]), - ], -) -def test_helius_retry_after_without_sleep_provider( - requests_mock, retry_after, min_delays -): - rpc_url = 'https://mainnet.helius-rpc.com/?api-key=test-key' - headers = {'Retry-After': retry_after} if retry_after is not None else {} - requests_mock.post( - rpc_url, - [ - {'status_code': 429, 'text': 'Too Many Requests', 'headers': headers}, - ] - * len(min_delays) - + [{'json': {'result': {'value': 1}}}], - ) - api = SolanaApi(base_url=rpc_url) - - with patch('time.sleep') as sleep, patch('time.time', return_value=1789023600): - result = api._request('getBalance', ['address']) - - assert result == {'result': {'value': 1}} - assert requests_mock.call_count == len(min_delays) + 1 - assert sleep.call_count == len(min_delays) - for sleep_call, min_delay in zip(sleep.call_args_list, min_delays): - assert min_delay <= sleep_call.args[0] <= min(min_delay + 0.25, 5) - assert all( - request.json() == requests_mock.request_history[0].json() - for request in requests_mock.request_history[1:] - ) - - -@pytest.mark.parametrize('retry_after', ['6', '60', 'Thu, 10 Sep 2026 07:01:00 GMT']) -def test_helius_long_retry_after_fails_without_early_retry(requests_mock, retry_after): - rpc_url = 'https://mainnet.helius-rpc.com/' - requests_mock.post( - rpc_url, - status_code=429, - text='Too Many Requests', - headers={'Retry-After': retry_after}, - ) - api = SolanaApi(base_url=rpc_url) - - with patch('time.sleep') as sleep, patch('time.time', return_value=1789023600): - with pytest.raises(ApiException, match='429'): - api.get_balance('address') - - assert requests_mock.call_count == 1 - sleep.assert_not_called() - - -@pytest.mark.parametrize('status_code', [400, 401, 403, 500, 503]) -def test_helius_does_not_retry_other_http_errors(requests_mock, status_code): - rpc_url = 'https://mainnet.helius-rpc.com/' - requests_mock.post(rpc_url, status_code=status_code, text='RPC unavailable') - api = SolanaApi(base_url=rpc_url) - - with patch('time.sleep') as sleep: - with pytest.raises(ApiException, match=str(status_code)): - api.get_balance('address') - - assert requests_mock.call_count == 1 - sleep.assert_not_called() - - -def test_helius_keeps_invalid_address_error(requests_mock): - rpc_url = 'https://mainnet.helius-rpc.com/' - requests_mock.post( - rpc_url, - json={'error': {'code': -32602, 'message': 'Invalid param: WrongSize'}}, - ) - api = SolanaApi(base_url=rpc_url) - - with patch('time.sleep') as sleep, pytest.raises(InvalidAddressException): - api.get_balance('invalid') - - assert requests_mock.call_count == 1 - sleep.assert_not_called() - - -@pytest.mark.parametrize( - 'rpc_url', ['https://proxy/solana/', 'https://helius-rpc.com.example.org/'] -) -def test_non_helius_throttling_keeps_existing_behavior(requests_mock, rpc_url): - requests_mock.post(rpc_url, status_code=429, text='Too Many Requests') - sleep_provider = FakeSleepProvider() - api = SolanaApi(base_url=rpc_url, sleep_provider=sleep_provider) - - with patch('time.sleep') as sleep, pytest.raises(ApiException, match='429'): - api.get_balance('address') - - assert requests_mock.call_count == 1 - assert sleep_provider.calls == [] - sleep.assert_not_called() - - -def test_helius_retries_staking_page_without_duplicate_accounts(requests_mock): - rpc_url = 'https://mainnet.helius-rpc.com/' - requests_mock.post( - rpc_url, - [ - { - 'json': { - 'result': { - 'accounts': [{'pubkey': 'first'}], - 'paginationKey': 'next', - } - } - }, - {'status_code': 429, 'text': 'Too Many Requests'}, - { - 'json': { - 'result': { - 'accounts': [{'pubkey': 'second'}], - 'paginationKey': None, - } - } - }, - ], - ) - api = SolanaApi(base_url=rpc_url) - - with patch('time.sleep'): - result = api._fetch_staked_sol('address') - - assert result['result'] == [{'pubkey': 'first'}, {'pubkey': 'second'}] - assert requests_mock.call_count == 3 - assert 'paginationKey' not in requests_mock.request_history[0].json()['params'][1] - retry_body = requests_mock.request_history[1].json() - assert retry_body['params'][1]['paginationKey'] == 'next' - assert requests_mock.request_history[2].json() == retry_body - - def test_build_coin_from_das_asset(): api = SolanaApi() asset = { diff --git a/blockapi/v2/api/solana.py b/blockapi/v2/api/solana.py index ab3f8b0..3986bd5 100644 --- a/blockapi/v2/api/solana.py +++ b/blockapi/v2/api/solana.py @@ -1,8 +1,5 @@ import json import logging -import random -import time -from email.utils import parsedate_to_datetime from typing import Optional, Union from urllib.parse import urlparse @@ -85,8 +82,6 @@ class SolanaApi(CustomizableBlockchainApi, BalanceMixin): STAKE_AUTHORITY_OFFSET = 44 HELIUS_RPC_DOMAIN = 'helius-rpc.com' HELIUS_PROGRAM_ACCOUNTS_PAGE_SIZE = 5000 - HELIUS_MAX_ATTEMPTS = 3 - HELIUS_MAX_RETRY_DELAY = 5 DAS_BATCH_SIZE = 1000 _JSONRPC_INVALID_PARAMS = -32602 @@ -336,7 +331,11 @@ def _fetch_staked_sol(self, address: str) -> dict: 'commitment': 'finalized', } - if not self._is_helius_rpc(): + hostname = urlparse(self.base_url).hostname or '' + is_helius_rpc = hostname == self.HELIUS_RPC_DOMAIN or hostname.endswith( + f'.{self.HELIUS_RPC_DOMAIN}' + ) + if not is_helius_rpc: return self._request( method='getProgramAccounts', params=[self.STAKE_PROGRAM_ID, config], @@ -431,46 +430,7 @@ def _request(self, method: str, params: Union[list, dict]) -> dict: 'params': params, } ) - headers = {'Content-Type': 'application/json'} - if not self._is_helius_rpc(): - return self.post(body=body, headers=headers) - - for attempt in range(self.HELIUS_MAX_ATTEMPTS): - response = self._session.post(self.base_url, data=body, headers=headers) - if response.status_code != 429 or attempt == self.HELIUS_MAX_ATTEMPTS - 1: - break - - delay = self._helius_retry_delay(response, attempt) - # Leave long cooldowns to the caller's next fetch instead of blocking it. - if delay > self.HELIUS_MAX_RETRY_DELAY: - break - delay = min(delay + random.uniform(0, 0.25), self.HELIUS_MAX_RETRY_DELAY) - logger.warning(f'Helius {method} rate limited; retrying in {delay:.2f}s') - response.close() - time.sleep(delay) - - return self._check_and_get_from_response(response) - - def _is_helius_rpc(self) -> bool: - hostname = urlparse(self.base_url).hostname or '' - return hostname == self.HELIUS_RPC_DOMAIN or hostname.endswith( - f'.{self.HELIUS_RPC_DOMAIN}' - ) - - @staticmethod - def _helius_retry_delay(response: Response, attempt: int) -> float: - delay = 2**attempt - retry_after = response.headers.get('Retry-After') - if retry_after: - try: - return max(delay, int(retry_after)) - except ValueError: - try: - retry_at = parsedate_to_datetime(retry_after).timestamp() - return max(delay, retry_at - time.time()) - except (TypeError, ValueError, OverflowError): - pass - return delay + return self.post(body=body, headers={'Content-Type': 'application/json'}) def _opt_raise_on_other_error(self, response: Response) -> None: """Raise ApiException or InvalidAddressException on RPC errors.""" From e08f8969ebe7cafb31a45637d27463d7fc185da7 Mon Sep 17 00:00:00 2001 From: Zbynek Masler Date: Fri, 11 Sep 2026 12:10:31 +0200 Subject: [PATCH 4/4] fix(solana): retry Helius 429 once per fetch after 1.1 seconds --- blockapi/test/v2/api/test_solana.py | 120 ++++++++++++++++++++++++---- blockapi/v2/api/solana.py | 68 +++++++++++++--- 2 files changed, 160 insertions(+), 28 deletions(-) diff --git a/blockapi/test/v2/api/test_solana.py b/blockapi/test/v2/api/test_solana.py index 8202dc7..e3a60a0 100644 --- a/blockapi/test/v2/api/test_solana.py +++ b/blockapi/test/v2/api/test_solana.py @@ -3,7 +3,7 @@ from unittest.mock import call, patch import pytest -from requests_mock import ANY, Mocker +from requests_mock import Mocker from blockapi.test.v2.api.conftest import read_file from blockapi.v2.api import SolanaApi, SolscanApi @@ -77,10 +77,41 @@ def test_use_base_url(): @pytest.mark.parametrize( - ('rpc_url', 'uses_v2_staking'), + ('rpc_url', 'status', 'retry_after'), [ - ('https://mainnet.helius-rpc.com/', True), - ('https://proxy/solana/', False), + ('https://proxy/solana/', 429, None), + ('https://not-helius-rpc.com/', 429, None), + ('https://helius-rpc.com.example.org/', 429, None), + ('https://mainnet.helius-rpc.com/', 400, None), + ('https://mainnet.helius-rpc.com/', 500, None), + ('https://mainnet.helius-rpc.com/', 429, '5'), + ('https://mainnet.helius-rpc.com/', 429, 'Fri, 31 Dec 9999 23:59:59 GMT'), + ('https://mainnet.helius-rpc.com/', 429, 'invalid'), + ], +) +def test_fetch_does_not_retry_other_errors_or_long_cooldowns( + rpc_url, status, retry_after +): + headers = {'Retry-After': retry_after} if retry_after is not None else {} + with Mocker() as m, patch('time.sleep') as sleep: + m.post(rpc_url, status_code=status, headers=headers) + with pytest.raises(ApiException): + SolanaApi(base_url=rpc_url).fetch_balances('address') + + assert len(m.request_history) == 1 + sleep.assert_not_called() + + +@pytest.mark.parametrize( + ('rpc_url', 'uses_v2_staking', 'retry_index', 'retry_after'), + [ + ('https://mainnet.helius-rpc.com/', True, None, None), + ('https://proxy/solana/', False, None, None), + ('https://mainnet.helius-rpc.com/', True, 0, None), + ('https://mainnet.helius-rpc.com/', True, 1, '0'), + ('https://mainnet.helius-rpc.com/', True, 2, '1'), + ('https://mainnet.helius-rpc.com/', True, 3, None), + ('https://mainnet.helius-rpc.com/', True, 5, None), ], ) def test_get_balance_supports_helius_and_legacy_staking_responses( @@ -90,6 +121,8 @@ def test_get_balance_supports_helius_and_legacy_staking_responses( staked_solana_response, rpc_url, uses_v2_staking, + retry_index, + retry_after, ): test_addr = '5PjMxaijeVVQtuEzxK2NxyJeWwUbpTsi2uXuZ653WoHu' empty_token_accounts = '{"jsonrpc":"2.0","result":{"context":{"apiVersion":"1.17.34","slot":268207149},"value":[]},"id":1}' @@ -97,26 +130,39 @@ def test_get_balance_supports_helius_and_legacy_staking_responses( if not uses_v2_staking: staking_response['result'] = staking_response['result']['accounts'] - iterator = iter( - [ + responses = [ + {'text': data} + for data in [ sol_balance_response, token_accounts_response, empty_token_accounts, das_asset_batch_response, json.dumps(staking_response), ] - ) - - def get_text(*args, **kwargs): - assert args[0].url == rpc_url - data = next(iterator) - return data - - with Mocker() as m: - m.post(ANY, text=get_text) + ] + if uses_v2_staking: + responses.insert( + 4, {'json': {'result': {'accounts': [], 'paginationKey': 'next-page'}}} + ) + if retry_index is not None: + headers = {'Retry-After': retry_after} if retry_after is not None else {} + responses.insert(retry_index, {'status_code': 429, 'headers': headers}) + + with Mocker() as m, patch('time.sleep') as sleep: + m.post(rpc_url, responses) api = SolanaApi(base_url=rpc_url) balances = api.get_balance(test_addr) + if retry_index is not None: + sleep.assert_called_once_with(1.1) + assert len(m.request_history) == len(responses) + assert ( + m.request_history[retry_index].body + == m.request_history[retry_index + 1].body + ) + else: + sleep.assert_not_called() + staking_balances = { balance.asset_type: balance.balance_raw for balance in balances @@ -128,6 +174,47 @@ def get_text(*args, **kwargs): } +@pytest.mark.parametrize('fail_on_same_rpc', [True, False]) +def test_retry_is_shared_across_fetch_and_resets_for_next_fetch( + token_accounts_response, fail_on_same_rpc +): + api = SolanaApi(base_url='https://mainnet.helius-rpc.com/') + responses = [{'status_code': 429}] + if not fail_on_same_rpc: + responses += [ + {'json': {'result': {'value': 0}}}, + {'text': token_accounts_response}, + {'json': {'result': {'value': []}}}, + {'status_code': 429}, # Optional DAS metadata may fail. + {'json': {'result': {'accounts': [], 'paginationKey': 'next-page'}}}, + ] + responses.append({'status_code': 429}) + + with Mocker() as m, patch('time.sleep') as sleep: + for attempt in range(2): + m.post(api.base_url, responses) + with pytest.raises(ApiException): + api.fetch_balances('address') + assert len(m.request_history) == (attempt + 1) * len(responses) + if not fail_on_same_rpc: + assert ( + m.last_request.json()['params'][1]['paginationKey'] == 'next-page' + ) + + assert sleep.call_args_list == [call(1.1), call(1.1)] + + +def test_get_coin_retries_helius_metadata_once(): + api = SolanaApi(base_url='https://mainnet.helius-rpc.com/') + with Mocker() as m, patch('time.sleep') as sleep: + m.post(api.base_url, [{'status_code': 429}, {'json': {'result': []}}]) + coin = api.get_coin(('mint', 6)) + + assert coin.address == 'mint' + assert len(m.request_history) == 2 + sleep.assert_called_once_with(1.1) + + def test_build_coin_from_das_asset(): api = SolanaApi() asset = { @@ -317,6 +404,7 @@ def test_fetch_staked_sol_uses_v2_pagination(): call( method='getProgramAccountsV2', params=[api.STAKE_PROGRAM_ID, config], + retry=None, ), call( method='getProgramAccountsV2', @@ -324,6 +412,7 @@ def test_fetch_staked_sol_uses_v2_pagination(): api.STAKE_PROGRAM_ID, {**config, 'paginationKey': 'next-page'}, ], + retry=None, ), call( method='getProgramAccountsV2', @@ -331,6 +420,7 @@ def test_fetch_staked_sol_uses_v2_pagination(): api.STAKE_PROGRAM_ID, {**config, 'paginationKey': 'last-page'}, ], + retry=None, ), ] assert response == { diff --git a/blockapi/v2/api/solana.py b/blockapi/v2/api/solana.py index 3986bd5..421e5a3 100644 --- a/blockapi/v2/api/solana.py +++ b/blockapi/v2/api/solana.py @@ -1,5 +1,7 @@ import json import logging +import time +from dataclasses import dataclass from typing import Optional, Union from urllib.parse import urlparse @@ -32,6 +34,11 @@ logger = logging.getLogger(__name__) +@dataclass +class _HeliusRetry: + used: bool = False + + class SolanaApi(CustomizableBlockchainApi, BalanceMixin): """Solana JSON-RPC client with DAS metadata integration. @@ -81,6 +88,7 @@ class SolanaApi(CustomizableBlockchainApi, BalanceMixin): STAKE_PROGRAM_ID = 'Stake11111111111111111111111111111111111111' STAKE_AUTHORITY_OFFSET = 44 HELIUS_RPC_DOMAIN = 'helius-rpc.com' + HELIUS_RETRY_DELAY = 1.1 HELIUS_PROGRAM_ACCOUNTS_PAGE_SIZE = 5000 DAS_BATCH_SIZE = 1000 _JSONRPC_INVALID_PARAMS = -32602 @@ -104,7 +112,8 @@ def __init__( def fetch_balances(self, address: str) -> FetchResult: """Fetch native SOL, token accounts, DAS metadata, and staking data.""" - sol_response = self._request('getBalance', [address]) + retry = _HeliusRetry() + sol_response = self._request('getBalance', [address], retry=retry) raw_token_balances = self._request( 'getTokenAccountsByOwner', @@ -113,6 +122,7 @@ def fetch_balances(self, address: str) -> FetchResult: {'programId': self.TOKEN_PROGRAM_ID}, {'encoding': 'jsonParsed'}, ], + retry=retry, ) raw_token2022_balances = self._request( 'getTokenAccountsByOwner', @@ -121,14 +131,15 @@ def fetch_balances(self, address: str) -> FetchResult: {'programId': self.TOKEN_2022_PROGRAM_ID}, {'encoding': 'jsonParsed'}, ], + retry=retry, ) mint_addresses = self._collect_mint_addresses( raw_token_balances, raw_token2022_balances ) - self._fetch_das_assets(mint_addresses) + self._fetch_das_assets(mint_addresses, retry=retry) - raw_staked_sol = self._fetch_staked_sol(address) + raw_staked_sol = self._fetch_staked_sol(address, retry=retry) return FetchResult( data=sol_response, @@ -172,7 +183,7 @@ def parse_balances(self, fetch_result: FetchResult) -> ParseResult: def get_coin(self, fetch_params: tuple[str, int]) -> Coin: """Fetch and build a Coin for a given contract address and decimals.""" contract, decimals = fetch_params - self._fetch_das_assets([contract]) + self._fetch_das_assets([contract], retry=_HeliusRetry()) return self._resolve_coin(contract, decimals) @staticmethod @@ -248,7 +259,9 @@ def _extract_token_info(account: dict) -> dict: # ── DAS integration ──────────────────────────────────────── - def _fetch_das_assets(self, mint_addresses: list[str]) -> None: + def _fetch_das_assets( + self, mint_addresses: list[str], retry: Optional[_HeliusRetry] = None + ) -> None: """Batch-fetch token metadata via DAS and populate cache.""" uncached = list( dict.fromkeys(m for m in mint_addresses if m not in self._das_cache) @@ -262,6 +275,7 @@ def _fetch_das_assets(self, mint_addresses: list[str]) -> None: response = self._request( 'getAssetBatch', {'ids': chunk, 'options': {'showFungible': True}}, + retry=retry, ) except (ApiException, RequestException) as e: logger.warning('DAS getAssetBatch failed: %s', e) @@ -315,7 +329,9 @@ def _build_coin_from_das_asset(self, asset: dict) -> Optional[Coin]: # ── Staking ──────────────────────────────────────────────── - def _fetch_staked_sol(self, address: str) -> dict: + def _fetch_staked_sol( + self, address: str, retry: Optional[_HeliusRetry] = None + ) -> dict: """Fetch staked SOL accounts for a given address.""" config = { 'filters': [ @@ -331,11 +347,7 @@ def _fetch_staked_sol(self, address: str) -> dict: 'commitment': 'finalized', } - hostname = urlparse(self.base_url).hostname or '' - is_helius_rpc = hostname == self.HELIUS_RPC_DOMAIN or hostname.endswith( - f'.{self.HELIUS_RPC_DOMAIN}' - ) - if not is_helius_rpc: + if not self._is_helius_rpc(): return self._request( method='getProgramAccounts', params=[self.STAKE_PROGRAM_ID, config], @@ -349,6 +361,7 @@ def _fetch_staked_sol(self, address: str) -> dict: response = self._request( method='getProgramAccountsV2', params=[self.STAKE_PROGRAM_ID, config], + retry=retry, ) page = response['result'] accounts.extend(page['accounts']) @@ -419,7 +432,12 @@ def _parse_rent_reserve( # ── Infrastructure ───────────────────────────────────────── - def _request(self, method: str, params: Union[list, dict]) -> dict: + def _request( + self, + method: str, + params: Union[list, dict], + retry: Optional[_HeliusRetry] = None, + ) -> dict: """Send a JSON-RPC request to the Solana RPC endpoint.""" self._request_id += 1 body = json.dumps( @@ -430,7 +448,31 @@ def _request(self, method: str, params: Union[list, dict]) -> dict: 'params': params, } ) - return self.post(body=body, headers={'Content-Type': 'application/json'}) + headers = {'Content-Type': 'application/json'} + if not self._is_helius_rpc(): + return self.post(body=body, headers=headers) + + response = self._session.post(self.base_url, data=body, headers=headers) + if ( + response.status_code == 429 + and retry is not None + and not retry.used + and response.headers.get('Retry-After', '0').strip() in ('0', '1') + ): + retry.used = True + response.close() + logger.warning( + f'Helius {method} rate limited; retrying in {self.HELIUS_RETRY_DELAY}s' + ) + time.sleep(self.HELIUS_RETRY_DELAY) + response = self._session.post(self.base_url, data=body, headers=headers) + return self._check_and_get_from_response(response) + + def _is_helius_rpc(self) -> bool: + hostname = urlparse(self.base_url).hostname or '' + return hostname == self.HELIUS_RPC_DOMAIN or hostname.endswith( + f'.{self.HELIUS_RPC_DOMAIN}' + ) def _opt_raise_on_other_error(self, response: Response) -> None: """Raise ApiException or InvalidAddressException on RPC errors."""