diff --git a/CHANGELOG.md b/CHANGELOG.md index 2f63e51..5c8f653 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,8 @@ All notable changes to the Aptos Python SDK will be captured in this file. This ## Unreleased +- Make e2e / localnet examples more reliable: `aggregator_value` reads both OptionalAggregator variants (aggregator table and integer, as used on localnet), REST and faucet calls retry transient 429/5xx and faucet sequence-number races, and the integration harness waits for the node and reloads network env vars after starting a localnet. + ## 0.12.0 (2026-07-02) ### Breaking changes diff --git a/aptos_sdk/async_client.py b/aptos_sdk/async_client.py index b266fcd..6ac3f0c 100644 --- a/aptos_sdk/async_client.py +++ b/aptos_sdk/async_client.py @@ -5,6 +5,7 @@ import json as _json import logging import time +from collections.abc import Awaitable, Callable from dataclasses import dataclass from typing import Any, Dict, List, Optional @@ -29,6 +30,26 @@ U64_MAX = 18446744073709551615 +# Transient REST / faucet failures that are safe to retry. SEQUENCE_NUMBER_* +# shows up when a faucet minter races concurrent mint transactions. +_FAUCET_RETRY_MARKERS = ( + "SEQUENCE_NUMBER_TOO_OLD", + "SEQUENCE_NUMBER_TOO_NEW", + "TRANSACTION_EXPIRED", +) + + +def _retryable_http_status(status: int) -> bool: + return status == 429 or status >= 500 + + +def _retryable_faucet_error(status: int, body: str) -> bool: + if _retryable_http_status(status): + return True + if status >= 400: + return any(marker in body for marker in _FAUCET_RETRY_MARKERS) + return False + @dataclass class ClientConfig: @@ -40,6 +61,7 @@ class ClientConfig: transaction_wait_in_seconds: int = 20 http2: bool = True api_key: Optional[str] = None + http_retries: int = 3 class IndexerClient: @@ -457,48 +479,73 @@ async def aggregator_value( resource_type: str, aggregator_path: List[str], ) -> int: + """Read an ``OptionalAggregator`` value from an account resource. + + Aptos ``0x1::optional_aggregator::OptionalAggregator`` stores either a + parallelizable aggregator (table handle + key) or a plain integer. Local + networks typically use the integer variant for APT supply; mainnet and + devnet use the aggregator variant. This helper accepts both so callers + do not need to know which representation the chain is using. + + :param account_address: Account that holds the resource. + :param resource_type: Move resource type, e.g. CoinInfo. + :param aggregator_path: Field names from the resource root to the + ``OptionalAggregator``, e.g. ``["supply"]``. The list is not mutated. + """ source = await self.account_resource(account_address, resource_type) source_data = data = source["data"] - while len(aggregator_path) > 0: - key = aggregator_path.pop() + for key in aggregator_path: if key not in data: raise ApiError(f"aggregator path not found in data: {source_data}", source_data) data = data[key] - if "vec" not in data: - raise ApiError(f"aggregator not found in data: {source_data}", source_data) - data = data["vec"] - if len(data) != 1: - raise ApiError(f"aggregator not found in data: {source_data}", source_data) - data = data[0] - if "aggregator" not in data: - raise ApiError(f"aggregator not found in data: {source_data}", source_data) - data = data["aggregator"] - if "vec" not in data: - raise ApiError(f"aggregator not found in data: {source_data}", source_data) - data = data["vec"] - if len(data) != 1: + if "vec" not in data or len(data["vec"]) != 1: raise ApiError(f"aggregator not found in data: {source_data}", source_data) - data = data[0] - if "handle" not in data: - raise ApiError(f"aggregator not found in data: {source_data}", source_data) - if "key" not in data: - raise ApiError(f"aggregator not found in data: {source_data}", source_data) - handle = data["handle"] - key = data["key"] - return int(await self.get_table_item(handle, "address", "u128", key)) + optional = data["vec"][0] + + aggregator = optional.get("aggregator", {}).get("vec", []) + if len(aggregator) == 1 and "handle" in aggregator[0] and "key" in aggregator[0]: + handle = aggregator[0]["handle"] + key = aggregator[0]["key"] + return int(await self.get_table_item(handle, "address", "u128", key)) + + integer = optional.get("integer", {}).get("vec", []) + if len(integer) == 1 and "value" in integer[0]: + return int(integer[0]["value"]) + + raise ApiError(f"aggregator not found in data: {source_data}", source_data) # # Ledger accessors # async def info(self) -> Dict[str, str]: - response = await self.client.get(self.base_url) + async def send() -> httpx.Response: + return await self.client.get(self.base_url) + + response = await self._send_with_retry(send) if response.status_code >= 400: raise ApiError(response.text, response.status_code) return response.json() + async def wait_until_ready(self, timeout_secs: float = 60.0) -> None: + """Poll ledger info until the node responds successfully. + + Used by integration tests to wait out localnet startup races instead of + failing on the first connection error. + """ + deadline = time.monotonic() + timeout_secs + last_error: Optional[Exception] = None + while time.monotonic() < deadline: + try: + await self.info() + return + except (ApiError, httpx.RequestError) as exc: + last_error = exc + await asyncio.sleep(0.5) + raise TimeoutError(f"node not ready after {timeout_secs}s: {last_error}") + # # Transactions # @@ -932,6 +979,32 @@ async def view_bcs_payload( raise ApiError(response.text, response.status_code) return response.json() + async def _send_with_retry( + self, send: Callable[[], Awaitable[httpx.Response]] + ) -> httpx.Response: + """Retry GETs and idempotent POSTs on transport errors, 429, and 5xx. + + Transaction submission must not use this helper: a lost 5xx response + after the node accepted the transaction would double-submit. + """ + retries = self.client_config.http_retries + last_response: Optional[httpx.Response] = None + for attempt in range(retries + 1): + try: + response = await send() + except httpx.RequestError: + if attempt < retries: + await asyncio.sleep(0.25 * (2**attempt)) + continue + raise + if _retryable_http_status(response.status_code) and attempt < retries: + last_response = response + await asyncio.sleep(0.25 * (2**attempt)) + continue + return response + assert last_response is not None + return last_response + async def _post( self, endpoint: str, @@ -942,21 +1015,29 @@ async def _post( # format params: params = {} if params is None else params params = {key: val for key, val in params.items() if val is not None} - return await self.client.post( - url=f"{self.base_url}/{endpoint}", - params=params, - headers=headers, - json=data, - ) + + async def send() -> httpx.Response: + return await self.client.post( + url=f"{self.base_url}/{endpoint}", + params=params, + headers=headers, + json=data, + ) + + return await self._send_with_retry(send) async def _get(self, endpoint: str, params: Optional[Dict[str, Any]] = None) -> httpx.Response: # format params: params = {} if params is None else params params = {key: val for key, val in params.items() if val is not None} - return await self.client.get( - url=f"{self.base_url}/{endpoint}", - params=params, - ) + + async def send() -> httpx.Response: + return await self.client.get( + url=f"{self.base_url}/{endpoint}", + params=params, + ) + + return await self._send_with_retry(send) class FaucetClient: @@ -969,6 +1050,7 @@ class FaucetClient: base_url: str rest_client: RestClient headers: Dict[str, str] + _fund_lock: asyncio.Lock def __init__(self, base_url: str, rest_client: RestClient, auth_token: Optional[str] = None): self.base_url = base_url @@ -976,6 +1058,7 @@ def __init__(self, base_url: str, rest_client: RestClient, auth_token: Optional[ self.headers = {"Content-Type": "application/json"} if auth_token: self.headers["Authorization"] = f"Bearer {auth_token}" + self._fund_lock = asyncio.Lock() async def close(self) -> None: """Close the underlying REST client connection.""" @@ -987,20 +1070,52 @@ async def fund_account( """This creates an account if it does not exist and mints the specified amount of coins into that account. + Concurrent calls on the same client are serialized so a single faucet + minter does not race sequence numbers. Transient errors (429, 5xx, + SEQUENCE_NUMBER_TOO_OLD/NEW) are retried. + Note: only devnet has a publicly accessible faucet. For testnet, you must initialize this client with an auth_token. """ - response = await self.rest_client.client.post( - f"{self.base_url}/fund", - headers=self.headers, - json={"address": str(address), "amount": amount}, - ) - if response.status_code >= 400: - raise ApiError(response.text, response.status_code) - txn_hash = response.json()["txn_hashes"][0] - if wait_for_transaction: - await self.rest_client.wait_for_transaction(txn_hash) - return txn_hash + async with self._fund_lock: + return await self._fund_account_once(address, amount, wait_for_transaction) + + async def _fund_account_once( + self, address: AccountAddress, amount: int, wait_for_transaction: bool + ) -> str: + retries = self.rest_client.client_config.http_retries + last_error: Optional[ApiError] = None + for attempt in range(retries + 1): + try: + response = await self.rest_client.client.post( + f"{self.base_url}/fund", + headers=self.headers, + json={"address": str(address), "amount": amount}, + ) + except httpx.RequestError as exc: + last_error = ApiError(str(exc), 0) + if attempt < retries: + await asyncio.sleep(0.25 * (2**attempt)) + continue + raise last_error from exc + + if response.status_code >= 400: + last_error = ApiError(response.text, response.status_code) + if ( + _retryable_faucet_error(response.status_code, response.text) + and attempt < retries + ): + await asyncio.sleep(0.25 * (2**attempt)) + continue + raise last_error + + txn_hash = response.json()["txn_hashes"][0] + if wait_for_transaction: + await self.rest_client.wait_for_transaction(txn_hash) + return txn_hash + + assert last_error is not None + raise last_error async def healthy(self) -> bool: """Return ``True`` iff the faucet's root endpoint reports ``tap:ok``.""" diff --git a/aptos_sdk/extra_test.py b/aptos_sdk/extra_test.py index 2647239..7607d7b 100644 --- a/aptos_sdk/extra_test.py +++ b/aptos_sdk/extra_test.py @@ -96,9 +96,13 @@ def _build_rest_client( handler: Callable[[httpx.Request], httpx.Response], *, base_url: str = "http://mock.invalid/v1", + http_retries: int = 0, ) -> RestClient: """Construct a RestClient whose underlying httpx.AsyncClient is mocked.""" - client = RestClient(base_url, ClientConfig(http2=False, transaction_wait_in_seconds=2)) + client = RestClient( + base_url, + ClientConfig(http2=False, transaction_wait_in_seconds=2, http_retries=http_retries), + ) _run(client.client.aclose()) transport = httpx.MockTransport(handler) client.client = httpx.AsyncClient(base_url="", transport=transport) @@ -453,14 +457,89 @@ def handler(req: httpx.Request) -> httpx.Response: return httpx.Response(404) client = _build_rest_client(handler) + path = ["supply"] v = _run( client.aggregator_value( AccountAddress.from_str_relaxed(SAMPLE_ADDR), "0x1::coin::CoinInfo", - ["supply"], + path, ) ) self.assertEqual(v, 42) + self.assertEqual(path, ["supply"]) + _run(client.close()) + + def test_aggregator_value_reads_integer_optional_aggregator(self): + """Localnet stores APT supply as OptionalAggregator.integer, not a table.""" + resource = { + "data": { + "decimals": 8, + "name": "Aptos Coin", + "supply": { + "vec": [ + { + "aggregator": {"vec": []}, + "integer": { + "vec": [ + { + "limit": "340282366920938463463374607431768211455", + "value": "43510", + } + ] + }, + } + ] + }, + "symbol": "APT", + } + } + + def handler(req: httpx.Request) -> httpx.Response: + if "/resource/" in req.url.path: + return httpx.Response(200, json=resource) + return httpx.Response(404) + + client = _build_rest_client(handler) + v = _run( + client.aggregator_value( + AccountAddress.from_str_relaxed(SAMPLE_ADDR), + "0x1::coin::CoinInfo<0x1::aptos_coin::AptosCoin>", + ["supply"], + ) + ) + self.assertEqual(v, 43510) + _run(client.close()) + + def test_aggregator_value_walks_path_outer_to_inner(self): + resource = { + "data": { + "outer": { + "inner": { + "vec": [ + { + "aggregator": {"vec": []}, + "integer": {"vec": [{"value": "7"}]}, + } + ] + } + } + } + } + + def handler(req: httpx.Request) -> httpx.Response: + if "/resource/" in req.url.path: + return httpx.Response(200, json=resource) + return httpx.Response(404) + + client = _build_rest_client(handler) + v = _run( + client.aggregator_value( + AccountAddress.from_str_relaxed(SAMPLE_ADDR), + "0x1::coin::CoinInfo", + ["outer", "inner"], + ) + ) + self.assertEqual(v, 7) _run(client.close()) def test_aggregator_value_missing_path_raises(self): @@ -527,6 +606,37 @@ def handler(req: httpx.Request) -> httpx.Response: _run(coro) _run(client.close()) + def test_get_retries_transient_503(self): + calls = {"n": 0} + + def handler(req: httpx.Request) -> httpx.Response: + calls["n"] += 1 + if calls["n"] == 1: + return httpx.Response(503, text="unavailable") + return httpx.Response(200, json={"chain_id": 4}) + + client = _build_rest_client(handler, http_retries=3) + with mock.patch("aptos_sdk.async_client.asyncio.sleep", new=mock.AsyncMock()): + info = _run(client.info()) + self.assertEqual(info["chain_id"], 4) + self.assertEqual(calls["n"], 2) + _run(client.close()) + + def test_wait_until_ready_polls_until_info_succeeds(self): + calls = {"n": 0} + + def handler(req: httpx.Request) -> httpx.Response: + calls["n"] += 1 + if calls["n"] == 1: + return httpx.Response(500, text="booting") + return httpx.Response(200, json={"chain_id": 4}) + + client = _build_rest_client(handler, http_retries=0) + with mock.patch("aptos_sdk.async_client.asyncio.sleep", new=mock.AsyncMock()): + _run(client.wait_until_ready(timeout_secs=5)) + self.assertGreaterEqual(calls["n"], 2) + _run(client.close()) + class FaucetClientTests(unittest.TestCase): def _client(self, handler): @@ -574,6 +684,29 @@ def handler(req: httpx.Request) -> httpx.Response: self.assertTrue(_run(c.healthy())) _run(c.close()) + def test_fund_account_retries_sequence_number_too_old(self): + calls = {"fund": 0} + + def handler(req: httpx.Request) -> httpx.Response: + if req.url.path.endswith("/fund"): + calls["fund"] += 1 + if calls["fund"] == 1: + return httpx.Response( + 400, + text="API error Error(VmError): Invalid transaction: " + "Type: Validation Code: SEQUENCE_NUMBER_TOO_OLD", + ) + return httpx.Response(200, json={"txn_hashes": ["0xabc"]}) + return httpx.Response(200, json={"type": "user_transaction", "success": True}) + + rest = _build_rest_client(handler, http_retries=3) + c = FaucetClient("http://faucet.invalid", rest, auth_token="tok") + with mock.patch("aptos_sdk.async_client.asyncio.sleep", new=mock.AsyncMock()): + h = _run(c.fund_account(AccountAddress.from_str_relaxed(SAMPLE_ADDR), 100)) + self.assertEqual(h, "0xabc") + self.assertEqual(calls["fund"], 2) + _run(c.close()) + class PackagePublisherTests(unittest.TestCase): def test_create_chunks_splits_input(self): diff --git a/examples/integration_test.py b/examples/integration_test.py index 2658e39..c249105 100644 --- a/examples/integration_test.py +++ b/examples/integration_test.py @@ -6,13 +6,16 @@ """ import asyncio +import importlib import os import unittest from typing import Optional from aptos_sdk.account_address import AccountAddress from aptos_sdk.aptos_cli_wrapper import AptosCLIWrapper, AptosInstance +from aptos_sdk.async_client import RestClient +from . import common from .common import APTOS_CORE_PATH @@ -22,6 +25,8 @@ class Test(unittest.IsolatedAsyncioTestCase): @classmethod def setUpClass(self): if os.getenv("APTOS_TEST_USE_EXISTING_NETWORK"): + self._reload_network_config() + asyncio.run(self._wait_for_network()) return self._node = AptosCLIWrapper.start_node() @@ -30,8 +35,23 @@ def setUpClass(self): raise Exception("".join(self._node.errors())) os.environ["APTOS_FAUCET_URL"] = "http://127.0.0.1:8081" - os.environ["APTOS_INDEXER_CLIENT"] = "none" + os.environ["APTOS_INDEXER_URL"] = "none" os.environ["APTOS_NODE_URL"] = "http://127.0.0.1:8080/v1" + self._reload_network_config() + asyncio.run(self._wait_for_network()) + + @classmethod + def _reload_network_config(cls) -> None: + """Re-read NODE/FAUCET/INDEXER URLs after env vars are set in setUpClass.""" + importlib.reload(common) + + @classmethod + async def _wait_for_network(cls) -> None: + rest = RestClient(common.NODE_URL, client_config=common.CLIENT_CONFIG) + try: + await rest.wait_until_ready() + finally: + await rest.close() async def test_aptos_token(self): from . import aptos_token diff --git a/examples/large_package_publisher.py b/examples/large_package_publisher.py index e0360f1..ff46a6a 100644 --- a/examples/large_package_publisher.py +++ b/examples/large_package_publisher.py @@ -54,10 +54,9 @@ async def main( faucet_client = FaucetClient(FAUCET_URL, rest_client, FAUCET_AUTH_TOKEN) alice = Account.generate() - req0 = faucet_client.fund_account(alice.address(), 1_000_000_000) - req1 = faucet_client.fund_account(alice.address(), 1_000_000_000) - req2 = faucet_client.fund_account(alice.address(), 1_000_000_000) - await asyncio.gather(*[req0, req1, req2]) + await faucet_client.fund_account(alice.address(), 1_000_000_000) + await faucet_client.fund_account(alice.address(), 1_000_000_000) + await faucet_client.fund_account(alice.address(), 1_000_000_000) alice_balance = await rest_client.account_balance(alice.address()) print(f"Alice: {alice.address()} {alice_balance}") diff --git a/examples/read_aggregator.py b/examples/read_aggregator.py index 632a8a5..083c9d6 100644 --- a/examples/read_aggregator.py +++ b/examples/read_aggregator.py @@ -11,6 +11,8 @@ async def main(): rest_client = RestClient(NODE_URL, client_config=CLIENT_CONFIG) + # CoinInfo.supply is an OptionalAggregator: parallelizable aggregator on + # long-lived networks, integer on localnet. aggregator_value handles both. total_apt = await rest_client.aggregator_value( AccountAddress.from_str("0x1"), "0x1::coin::CoinInfo<0x1::aptos_coin::AptosCoin>", diff --git a/examples/transfer_coin.py b/examples/transfer_coin.py index 07ae501..410ce98 100644 --- a/examples/transfer_coin.py +++ b/examples/transfer_coin.py @@ -79,7 +79,7 @@ async def main(): variables = {"account": f"{bob.address()}"} data = None last_error: Optional[Exception] = None - for _ in range(20): + for _ in range(30): try: data = await indexer_client.query(query, variables) except IndexerError as exc: