diff --git a/docs/api.rst b/docs/api.rst index cd16ff3..35939db 100644 --- a/docs/api.rst +++ b/docs/api.rst @@ -25,12 +25,15 @@ Providers .. autoclass:: HTTPProvider :show-inheritance: +.. autoclass:: ProviderPath + Fallback providers ------------------ .. autoclass:: FallbackProvider :show-inheritance: + :members: errors .. autoclass:: CycleFallback :show-inheritance: @@ -148,7 +151,7 @@ Testing utilities :show-inheritance: :members: disable_auto_mine_transactions, enable_auto_mine_transactions, take_snapshot, revert_to_snapshot -.. autoclass:: SnapshotID +.. autoclass:: SnapshotID() .. autoclass:: HTTPProviderServer :members: @@ -266,11 +269,11 @@ Compiled and deployed contracts Filter objects -------------- -.. autoclass:: pons._client_rpc.BlockFilter() +.. autoclass:: pons.BlockFilter() -.. autoclass:: pons._client_rpc.PendingTransactionFilter() +.. autoclass:: pons.PendingTransactionFilter() -.. autoclass:: pons._client_rpc.LogFilter() +.. autoclass:: pons.LogFilter() Solidity types diff --git a/docs/changelog.rst b/docs/changelog.rst index d2f8ef7..f6b5bcd 100644 --- a/docs/changelog.rst +++ b/docs/changelog.rst @@ -12,6 +12,8 @@ Changed - Renamed ``id_`` fields of ``BlockFilter``, ``PendingTransactionFilter``, and ``LogFilter`` to just ``id``. (PR_82_) - Split out ``http-provider-server`` feature from ``local-provider``. (PR_82_) - ``RemoteError`` removed. (PR_82_) +- The internal structure of ``BlockFilter``, ``LogFilter``, and ``PendingTransactionFilter`` is now undocumented. (PR_87_) +- The path returned by ``Provider.rpc_and_pin()`` is now a ``ProviderPath`` object. (PR_87_) Added @@ -23,12 +25,15 @@ Added - Exporting ``HTTPError`` and ``BadResponseFormat``. (PR_82_) - Exporting ``InvalidResponse``. (PR_86_) - ``LocalProvider.root`` is now of type ``AccountSigner`` instead of ``Signer``. (PR_86_) +- ``FallbackProvider`` now records encountered errors, and they can be accessed via the ``errors()`` method. (PR_87_) +- ``BlockFilter``, ``LogFilter``, and ``PendingTransactionFilter`` are exported from the top level. (PR_87_) .. _PR_81: https://github.com/fjarri-eth/pons/pull/81 .. _PR_82: https://github.com/fjarri-eth/pons/pull/82 .. _PR_84: https://github.com/fjarri-eth/pons/pull/84 .. _PR_86: https://github.com/fjarri-eth/pons/pull/86 +.. _PR_87: https://github.com/fjarri-eth/pons/pull/87 0.8.1 (2024-11-12) diff --git a/docs/conf.py b/docs/conf.py index 40237b8..ae6f2df 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -47,6 +47,11 @@ autoclass_content = "both" autodoc_member_order = "groupwise" +autodoc_default_options = { + #'members': True, + "undoc-members": False, + #'show-inheritance': True, +} # Add any paths that contain templates here, relative to this directory. templates_path = [] diff --git a/pons/__init__.py b/pons/__init__.py index acb6fa7..8483427 100644 --- a/pons/__init__.py +++ b/pons/__init__.py @@ -10,7 +10,14 @@ ContractPanic, TransactionFailed, ) -from ._client_rpc import BadResponseFormat, ClientSessionRPC, ProviderError +from ._client_rpc import ( + BadResponseFormat, + BlockFilter, + ClientSessionRPC, + LogFilter, + PendingTransactionFilter, + ProviderError, +) from ._compiler import EVMVersion, compile_contract_file from ._contract import ( BaseBoundMethodCall, @@ -54,6 +61,7 @@ InvalidResponse, ProtocolError, Provider, + ProviderPath, Unreachable, ) from ._signer import AccountSigner, Signer @@ -65,6 +73,7 @@ "AccountSigner", "BadResponseFormat", "BaseBoundMethodCall", + "BlockFilter", "BoundConstructor", "BoundConstructorCall", "BoundEvent", @@ -99,15 +108,18 @@ "HTTPProviderServer", "InvalidResponse", "LocalProvider", + "LogFilter", "Method", "MethodCall", "MultiMethod", "Multicall", "Mutability", + "PendingTransactionFilter", "PriorityFallback", "ProtocolError", "Provider", "ProviderError", + "ProviderPath", "Receive", "Signer", "SnapshotID", diff --git a/pons/_client_rpc.py b/pons/_client_rpc.py index 9ede0f1..f7f9a10 100644 --- a/pons/_client_rpc.py +++ b/pons/_client_rpc.py @@ -26,25 +26,43 @@ from ._contract import BaseBoundMethodCall from ._contract_abi import EventFilter -from ._provider import InvalidResponse, ProviderSession +from ._provider import InvalidResponse, ProviderPath, ProviderSession @dataclass class BlockFilter: + """ + A block filter created on a remote provider. + + Expires after some time subject to the provider's settings. + """ + id: int - provider_path: tuple[int, ...] + provider_path: ProviderPath @dataclass class PendingTransactionFilter: + """ + A pending transaction filter created on a remote provider. + + Expires after some time subject to the provider's settings. + """ + id: int - provider_path: tuple[int, ...] + provider_path: ProviderPath @dataclass class LogFilter: + """ + A log filter created on a remote provider. + + Expires after some time subject to the provider's settings. + """ + id: int - provider_path: tuple[int, ...] + provider_path: ProviderPath class BadResponseFormat(Exception): @@ -116,7 +134,7 @@ async def rpc_call( async def rpc_call_pin( provider_session: ProviderSession, method_name: str, ret_type: type[RetType], *args: Any -) -> tuple[RetType, tuple[int, ...]]: +) -> tuple[RetType, ProviderPath]: """Catches various response formatting errors and returns them in a unified way.""" with convert_errors(method_name): result, provider_path = await provider_session.rpc_and_pin( @@ -127,7 +145,7 @@ async def rpc_call_pin( async def rpc_call_at_pin( provider_session: ProviderSession, - provider_path: tuple[int, ...], + provider_path: ProviderPath, method_name: str, ret_type: type[RetType], *args: Any, diff --git a/pons/_fallback_provider.py b/pons/_fallback_provider.py index c3af8d7..2f6183a 100644 --- a/pons/_fallback_provider.py +++ b/pons/_fallback_provider.py @@ -1,17 +1,27 @@ +import threading from abc import ABC, abstractmethod -from collections.abc import AsyncIterator, Iterable +from collections.abc import AsyncIterator, Iterable, Mapping +from collections.abc import Set as AbstractSet from contextlib import AsyncExitStack, asynccontextmanager from ethereum_rpc import RPCError -from ._provider import RPC_JSON, InvalidResponse, Provider, ProviderSession +from ._provider import ( + RPC_JSON, + InvalidResponse, + ProtocolError, + Provider, + ProviderPath, + ProviderSession, + Unreachable, +) class FallbackStrategy(ABC): """An abstract class defining a fallback strategy for multiple providers.""" @abstractmethod - def get_provider_order(self) -> list[int]: + def get_provider_order(self) -> list[str]: """ Returns the suggested order of providers to query, based on the accumulated data. This method is called once on every high-level request to the provider. @@ -26,24 +36,23 @@ class FallbackStrategyFactory(ABC): """ @abstractmethod - def make_strategy(self, num_providers: int) -> FallbackStrategy: + def make_strategy(self, provider_ids: AbstractSet[str]) -> FallbackStrategy: """Returns a strategy object.""" class CycleFallbackStrategy(FallbackStrategy): - def __init__(self, weights: list[int]): - self._providers = list(range(len(weights))) + def __init__(self, weights: dict[str, int]): self._weights = weights + self._provider_ids = list(weights.keys()) self._counter = 0 - def get_provider_order(self) -> list[int]: - if self._counter == self._weights[0]: + def get_provider_order(self) -> list[str]: + if self._counter == self._weights[self._provider_ids[0]]: self._counter = 0 - self._providers = [*self._providers[1:], self._providers[0]] - self._weights = [*self._weights[1:], self._weights[0]] + self._provider_ids = [*self._provider_ids[1:], self._provider_ids[0]] self._counter += 1 - return list(self._providers) + return self._provider_ids class CycleFallback(FallbackStrategyFactory): @@ -62,7 +71,8 @@ def __init__(self, weights: None | Iterable[int] = None): else: self._weights = None - def make_strategy(self, num_providers: int) -> CycleFallbackStrategy: + def make_strategy(self, provider_ids: AbstractSet[str]) -> CycleFallbackStrategy: + num_providers = len(provider_ids) weights = self._weights or [1] * num_providers if len(weights) != num_providers: @@ -71,7 +81,7 @@ def make_strategy(self, num_providers: int) -> CycleFallbackStrategy: f"inconsistent with the number of providers ({num_providers})" ) - return CycleFallbackStrategy(weights) + return CycleFallbackStrategy(dict(zip(provider_ids, weights, strict=True))) class PriorityFallbackStrategy(FallbackStrategy): @@ -80,16 +90,16 @@ class PriorityFallbackStrategy(FallbackStrategy): they were given to ``FallbackProvider``, until a successful response is received. """ - def __init__(self, num_providers: int): - self._providers = list(range(num_providers)) + def __init__(self, provider_ids: AbstractSet[str]): + self._provider_ids = list(provider_ids) - def get_provider_order(self) -> list[int]: - return self._providers + def get_provider_order(self) -> list[str]: + return self._provider_ids class PriorityFallback(FallbackStrategyFactory): - def make_strategy(self, num_providers: int) -> PriorityFallbackStrategy: - return PriorityFallbackStrategy(num_providers) + def make_strategy(self, provider_ids: AbstractSet[str]) -> PriorityFallbackStrategy: + return PriorityFallbackStrategy(provider_ids) class FallbackProvider(Provider): @@ -102,80 +112,118 @@ class FallbackProvider(Provider): pointing to the same physical provider, for the purpose of stateful requests (e.g. filter creation). - If all requests finished with an error, the most informative error is raised. + If ``strategy`` is ``None``, an instance of :py:class:`PriorityFallback` is used. + + If a request attempt results in an error for which ``use_fallback`` returns ``True``, + the next provider based on the chosen strategy will be selected. + Otherwise (or if it is the last provider), the error is raised normally. """ def __init__( self, - providers: Iterable[Provider], - strategy: FallbackStrategyFactory = PriorityFallback(), + providers: Mapping[str, Provider], + strategy: FallbackStrategyFactory | None = None, *, same_provider: bool = False, ): - self._providers = list(providers) - self._strategy = strategy.make_strategy(len(self._providers)) + self._providers = dict(providers) + strategy_ = strategy if strategy is not None else PriorityFallback() + self._strategy = strategy_.make_strategy(self._providers.keys()) self._same_provider = same_provider + self._errors: dict[str, Exception] = {} + self._lock = threading.Lock() @asynccontextmanager async def session(self) -> AsyncIterator["ProviderSession"]: async with AsyncExitStack() as stack: - sessions = [ - await stack.enter_async_context(provider.session()) for provider in self._providers - ] + sessions = { + provider_id: await stack.enter_async_context(provider.session()) + for provider_id, provider in self._providers.items() + } yield FallbackProviderSession( - sessions, self._strategy, same_provider=self._same_provider + self, + sessions, + self._strategy, + same_provider=self._same_provider, ) + def errors(self) -> list[tuple[ProviderPath, Exception]]: + """ + Returns the list of recorded errors for sub-providers. + + Only the most recent error for every sub-provider is recorded. + Querying this method clears the recorded errors. + """ + errors = [] + + with self._lock: + for provider_id, error in self._errors.items(): + errors.append((ProviderPath([provider_id]), error)) + self._errors = {} + + for provider_id, provider in self._providers.items(): + if isinstance(provider, FallbackProvider): + sub_errors = provider.errors() + for sub_path, error in sub_errors: + errors.append((sub_path.group(provider_id), error)) + + return errors + + def record_error(self, provider_id: str, exc: Exception) -> None: + with self._lock: + self._errors[provider_id] = exc + class FallbackProviderSession(ProviderSession): def __init__( - self, sessions: list[ProviderSession], strategy: FallbackStrategy, *, same_provider: bool + self, + provider: FallbackProvider, + sessions: dict[str, ProviderSession], + strategy: FallbackStrategy, + *, + same_provider: bool, ): + self._provider = provider self._sessions = sessions self._strategy = strategy self._same_provider = same_provider - async def rpc_and_pin(self, method: str, *args: RPC_JSON) -> tuple[RPC_JSON, tuple[int, ...]]: - exceptions: list[Exception] = [] - provider_idxs = self._strategy.get_provider_order() - for provider_idx in provider_idxs: + async def rpc_and_pin(self, method: str, *args: RPC_JSON) -> tuple[RPC_JSON, ProviderPath]: + provider_ids = self._strategy.get_provider_order() + + for i, provider_id in enumerate(provider_ids): + session = self._sessions[provider_id] try: - result, sub_idx = await self._sessions[provider_idx].rpc_and_pin(method, *args) - # PERF203: There won't be a lot of providers, and we need to collect errors from each. - # BLE001: it's just a middleware, collecting all errors. - except Exception as exc: # noqa: PERF203, BLE001 - exceptions.append(exc) - else: - return result, (provider_idx, *sub_idx) - - # Here we may have a list with each element being - # `RPCError`, `ProtocolError`, `InvalidResponse`, or `Unreachable`. - # Since the users of `Provider` rely on the error being one of these types, - # we can only raise one. So we raise the one with the most information. - # - # RPC errors give the most information, since they usually signify our request was invalid, - # so all the providers would respond to it in the same manner. - # - # `InvalidResponse` means that the library is unable to parse the response for some reason, - # probably because of a bug. So it will be the same for all providers. - # - # The other two, `ProtocolError` and `Unreachable` is exactly why we have the fallback. - # It is pretty much expected to happen. - rpc_errors = [exc for exc in exceptions if isinstance(exc, RPCError)] - if len(rpc_errors) > 0: - raise rpc_errors[0] - invalid_responses = [exc for exc in exceptions if isinstance(exc, InvalidResponse)] - if len(invalid_responses) > 0: - raise invalid_responses[0] - raise exceptions[0] + result, sub_path = await session.rpc_and_pin(method, *args) + # These are the exceptions `Provider` can raise for a remote problem + except (RPCError, Unreachable, InvalidResponse, ProtocolError) as exc: + if not isinstance(session, FallbackProviderSession): + self._provider.record_error(provider_id, exc) + + if i < len(provider_ids) - 1: + continue + + raise + + break + + else: # pragma: no cover + # This branch will never be reached, because the loop will either return, + # or raise an exception. + raise NotImplementedError + + return result, sub_path.group(provider_id) async def rpc(self, method: str, *args: RPC_JSON) -> RPC_JSON: - result, _provider = await self.rpc_and_pin(method, *args) + result, _path = await self.rpc_and_pin(method, *args) return result - async def rpc_at_pin(self, path: tuple[int, ...], method: str, *args: RPC_JSON) -> RPC_JSON: + async def rpc_at_pin(self, path: ProviderPath, method: str, *args: RPC_JSON) -> RPC_JSON: if self._same_provider: return await self.rpc(method, *args) - if not path or path[0] < 0 or path[0] >= len(self._sessions): - raise ValueError(f"Invalid provider path: {path}") - return await self._sessions[path[0]].rpc_at_pin(path[1:], method, *args) + if path.is_empty(): + raise ValueError("Expected a non-empty provider path") + provider_id, sub_path = path.ungroup() + if provider_id not in self._sessions: + raise ValueError(f"Provider id `{provider_id}` not found") + return await self._sessions[provider_id].rpc_at_pin(sub_path, method, *args) diff --git a/pons/_provider.py b/pons/_provider.py index 9ac9f66..20d08e4 100644 --- a/pons/_provider.py +++ b/pons/_provider.py @@ -1,5 +1,5 @@ from abc import ABC, abstractmethod -from collections.abc import AsyncIterator, Mapping, Sequence +from collections.abc import AsyncIterator, Iterable, Mapping, Sequence from contextlib import asynccontextmanager from http import HTTPStatus from json import JSONDecodeError @@ -57,6 +57,42 @@ def __str__(self) -> str: return f"HTTP status {self.status}: {self.message}" +class ProviderPath: + """Identifies a pinned provider.""" + + def __init__(self, path: Iterable[str]): + self._path = tuple(path) + + def group(self, id_: str) -> "ProviderPath": + """Prepends ``id_`` to the path.""" + return ProviderPath((id_, *self._path)) + + def ungroup(self) -> "tuple[str, ProviderPath]": + """Returns the top-level id and the subpath.""" + return (self._path[0], ProviderPath(self._path[1:])) + + @classmethod + def empty(cls) -> "ProviderPath": + """Returns an empty path.""" + return cls(()) + + def is_empty(self) -> bool: + """Returns ``True`` if the path is empty.""" + return not bool(self._path) + + def __str__(self) -> str: + return "/".join(self._path) + + def __repr__(self) -> str: + return f"ProviderPath({self._path!r})" + + def __eq__(self, other: object) -> bool: + return isinstance(other, ProviderPath) and self._path == other._path + + def __hash__(self) -> int: + return hash((ProviderPath, self._path)) + + class Provider(ABC): """The base class for JSON RPC providers.""" @@ -88,21 +124,21 @@ async def rpc(self, method: str, *args: RPC_JSON) -> RPC_JSON: """Calls the given RPC method with the already json-ified arguments.""" ... - async def rpc_and_pin(self, method: str, *args: RPC_JSON) -> tuple[RPC_JSON, tuple[int, ...]]: + async def rpc_and_pin(self, method: str, *args: RPC_JSON) -> tuple[RPC_JSON, ProviderPath]: """ Calls the given RPC method and returns the path to the provider it succeded on. This method will be typically overriden by multi-provider implementations. """ - return await self.rpc(method, *args), () + return await self.rpc(method, *args), ProviderPath.empty() - async def rpc_at_pin(self, path: tuple[int, ...], method: str, *args: RPC_JSON) -> RPC_JSON: + async def rpc_at_pin(self, path: ProviderPath, method: str, *args: RPC_JSON) -> RPC_JSON: """ Calls the given RPC method at the provider by the given path (obtained previously from ``rpc_and_pin()``). This method will be typically overriden by multi-provider implementations. """ - if path != (): - raise ValueError(f"Unexpected provider path: {path}") + if not path.is_empty(): + raise ValueError(f"Expected an empty provider path, got: `{path}`") return await self.rpc(method, *args) diff --git a/tests/test_fallback_provider.py b/tests/test_fallback_provider.py index c73cc5d..064945a 100644 --- a/tests/test_fallback_provider.py +++ b/tests/test_fallback_provider.py @@ -2,13 +2,21 @@ from collections.abc import AsyncIterator from contextlib import asynccontextmanager from enum import Enum +from http import HTTPStatus import pytest from ethereum_rpc import ErrorCode, RPCError -from pons import CycleFallback, FallbackProvider, PriorityFallback, Unreachable +from pons import ( + CycleFallback, + FallbackProvider, + HTTPError, + PriorityFallback, + ProtocolError, + Unreachable, +) from pons._fallback_provider import PriorityFallbackStrategy -from pons._provider import RPC_JSON, InvalidResponse, Provider, ProviderSession +from pons._provider import RPC_JSON, InvalidResponse, Provider, ProviderPath, ProviderSession def random_request() -> str: @@ -20,6 +28,7 @@ class ProviderState(Enum): UNREACHABLE = 2 BAD_RESPONSE = 3 RPC_ERROR = 4 + PROTOCOL_ERROR = 5 class MockProvider(Provider): @@ -47,17 +56,20 @@ async def rpc(self, method: str, *_args: RPC_JSON) -> RPC_JSON: raise InvalidResponse("") if self.provider.state == ProviderState.RPC_ERROR: raise RPCError(ErrorCode(-1), "") + if self.provider.state == ProviderState.PROTOCOL_ERROR: + raise HTTPError(HTTPStatus(500), "") + return "success" async def test_default_fallback() -> None: - providers = [MockProvider() for i in range(3)] + providers = {str(i): MockProvider() for i in range(3)} provider = FallbackProvider(providers) assert isinstance(provider._strategy, PriorityFallbackStrategy) def test_inconsistent_weights_length() -> None: - providers = [MockProvider() for i in range(3)] + providers = {str(i): MockProvider() for i in range(3)} msg = r"Length of the weights \(2\) inconsistent with the number of providers \(3\)" with pytest.raises(ValueError, match=msg): FallbackProvider(providers, CycleFallback([1, 2])) @@ -65,7 +77,7 @@ def test_inconsistent_weights_length() -> None: async def test_cycle_fallback() -> None: strategy = CycleFallback() - providers = [MockProvider() for i in range(3)] + providers = {str(i): MockProvider() for i in range(3)} provider = FallbackProvider(providers, strategy) requests = [str(x) for x in range(10)] @@ -73,14 +85,14 @@ async def test_cycle_fallback() -> None: for request in requests: await session.rpc(request) - assert providers[0].requests == ["0", "3", "6", "9"] - assert providers[1].requests == ["1", "4", "7"] - assert providers[2].requests == ["2", "5", "8"] + assert providers["0"].requests == ["0", "3", "6", "9"] + assert providers["1"].requests == ["1", "4", "7"] + assert providers["2"].requests == ["2", "5", "8"] async def test_cycle_fallback_custom_weights() -> None: strategy = CycleFallback([3, 1, 2]) - providers = [MockProvider() for i in range(3)] + providers = {str(i): MockProvider() for i in range(3)} provider = FallbackProvider(providers, strategy) requests = [str(x) for x in range(10)] @@ -88,14 +100,14 @@ async def test_cycle_fallback_custom_weights() -> None: for request in requests: await session.rpc(request) - assert providers[0].requests == ["0", "1", "2", "6", "7", "8"] - assert providers[1].requests == ["3", "9"] - assert providers[2].requests == ["4", "5"] + assert providers["0"].requests == ["0", "1", "2", "6", "7", "8"] + assert providers["1"].requests == ["3", "9"] + assert providers["2"].requests == ["4", "5"] async def test_fallback_on_errors() -> None: strategy = PriorityFallback() - providers = [MockProvider() for i in range(3)] + providers = {str(i): MockProvider() for i in range(3)} provider = FallbackProvider(providers, strategy) async with provider.session() as session: @@ -103,107 +115,182 @@ async def test_fallback_on_errors() -> None: request = random_request() result = await session.rpc(request) assert result == "success" - assert providers[0].requests[-1] == request + assert providers["0"].requests[-1] == request # 0 is unreachable, 1 is queried next. - providers[0].set_state(ProviderState.UNREACHABLE) + providers["0"].set_state(ProviderState.UNREACHABLE) request = random_request() result = await session.rpc(request) assert result == "success" - assert providers[1].requests[-1] == request + assert providers["1"].requests[-1] == request - # 0 returns an RPC error, 1 is unreachable, 2 is queried next. - providers[0].set_state(ProviderState.RPC_ERROR) - providers[1].set_state(ProviderState.UNREACHABLE) + # 0 returns a protocol error, 1 is unreachable, 2 is queried next. + providers["0"].set_state(ProviderState.PROTOCOL_ERROR) + providers["1"].set_state(ProviderState.UNREACHABLE) request = random_request() result = await session.rpc(request) assert result == "success" - assert providers[2].requests[-1] == request + assert providers["2"].requests[-1] == request async def test_raising_errors() -> None: strategy = PriorityFallback() - providers = [MockProvider() for i in range(3)] + providers = {str(i): MockProvider() for i in range(3)} provider = FallbackProvider(providers, strategy) async with provider.session() as session: - # All are unreachable, an Unreachable is raised - providers[0].set_state(ProviderState.UNREACHABLE) - providers[1].set_state(ProviderState.UNREACHABLE) - providers[2].set_state(ProviderState.UNREACHABLE) + providers["0"].set_state(ProviderState.UNREACHABLE) + providers["1"].set_state(ProviderState.UNREACHABLE) + providers["2"].set_state(ProviderState.UNREACHABLE) with pytest.raises(Unreachable): await session.rpc(random_request()) - # InvalidResponse is raised if present - providers[0].set_state(ProviderState.UNREACHABLE) - providers[1].set_state(ProviderState.BAD_RESPONSE) - providers[2].set_state(ProviderState.UNREACHABLE) - with pytest.raises(InvalidResponse): + providers["0"].set_state(ProviderState.UNREACHABLE) + providers["1"].set_state(ProviderState.BAD_RESPONSE) + providers["2"].set_state(ProviderState.UNREACHABLE) + with pytest.raises(Unreachable): await session.rpc(random_request()) - # RPCError is raised if present - providers[0].set_state(ProviderState.UNREACHABLE) - providers[1].set_state(ProviderState.BAD_RESPONSE) - providers[2].set_state(ProviderState.RPC_ERROR) - with pytest.raises(RPCError): + providers["0"].set_state(ProviderState.UNREACHABLE) + providers["1"].set_state(ProviderState.RPC_ERROR) + providers["2"].set_state(ProviderState.BAD_RESPONSE) + with pytest.raises(InvalidResponse): await session.rpc(random_request()) async def test_nested_providers() -> None: - providers = [MockProvider() for i in range(4)] - subprovider1 = FallbackProvider([providers[0], providers[1]], PriorityFallback()) - subprovider2 = FallbackProvider( - [providers[2], providers[3]], PriorityFallback(), same_provider=True + providers = {str(i): MockProvider() for i in range(4)} + subprovider0 = FallbackProvider({"0": providers["0"], "1": providers["1"]}, PriorityFallback()) + subprovider1 = FallbackProvider( + {"2": providers["2"], "3": providers["3"]}, PriorityFallback(), same_provider=True ) - provider = FallbackProvider([subprovider1, subprovider2], PriorityFallback()) + provider = FallbackProvider({"s0": subprovider0, "s1": subprovider1}, PriorityFallback()) async with provider.session() as session: # All providers operational, provider 0 is pinned request = random_request() _result, path = await session.rpc_and_pin(request) - assert path == (0, 0) - assert providers[0].requests[-1] == request + assert str(path) == "s0/0" + assert providers["0"].requests[-1] == request # Provider 0 offline, so trying to rpc it specifically results in an error - providers[0].set_state(ProviderState.UNREACHABLE) + providers["0"].set_state(ProviderState.UNREACHABLE) with pytest.raises(Unreachable): await session.rpc_at_pin(path, request) # Provider 0 is still offline, pinning results in using provider 1 request = random_request() _result, path = await session.rpc_and_pin(request) - assert path == (0, 1) - assert providers[1].requests[-1] == request + assert str(path) == "s0/1" + assert providers["1"].requests[-1] == request request = random_request() _result = await session.rpc_at_pin(path, request) - assert providers[1].requests[-1] == request + assert providers["1"].requests[-1] == request # All but provider 2 are offline request = random_request() - providers[0].set_state(ProviderState.UNREACHABLE) - providers[1].set_state(ProviderState.UNREACHABLE) - providers[3].set_state(ProviderState.UNREACHABLE) + providers["0"].set_state(ProviderState.UNREACHABLE) + providers["1"].set_state(ProviderState.UNREACHABLE) + providers["3"].set_state(ProviderState.UNREACHABLE) _result, path = await session.rpc_and_pin(request) - assert path == (1, 0) - assert providers[2].requests[-1] == request + assert str(path) == "s1/2" + assert providers["2"].requests[-1] == request # Since provider 2 and 3 are marked as "same provider", # provider 3 can be used instead of the pinned provider 2 request = random_request() - providers[3].set_state(ProviderState.NORMAL) - providers[2].set_state(ProviderState.UNREACHABLE) + providers["3"].set_state(ProviderState.NORMAL) + providers["2"].set_state(ProviderState.UNREACHABLE) await session.rpc_at_pin(path, request) - assert providers[3].requests[-1] == request + assert providers["3"].requests[-1] == request async def test_invalid_path() -> None: - providers = [MockProvider() for i in range(4)] - subprovider1 = FallbackProvider([providers[0], providers[1]], PriorityFallback()) - subprovider2 = FallbackProvider( - [providers[2], providers[3]], PriorityFallback(), same_provider=True + subprovider0 = FallbackProvider({"0": MockProvider(), "1": MockProvider()}, PriorityFallback()) + subprovider1 = FallbackProvider( + {"2": MockProvider(), "3": MockProvider()}, PriorityFallback(), same_provider=True + ) + provider = FallbackProvider({"s0": subprovider0, "s1": subprovider1}, PriorityFallback()) + async with provider.session() as session: + with pytest.raises(ValueError, match="Provider id `s2` not found"): + await session.rpc_at_pin(ProviderPath(["s2", "0"]), random_request()) + + async with provider.session() as session: + with pytest.raises(ValueError, match="Expected a non-empty provider path"): + await session.rpc_at_pin(ProviderPath.empty(), random_request()) + + +def assert_errors( + errors: list[tuple[ProviderPath, Exception]], + reference: list[tuple[ProviderPath, type[Exception]]], +) -> None: + # Since exceptions don't implement equality + for (test_path, exc), (ref_path, exc_type) in zip(errors, reference, strict=True): + assert test_path == ref_path + assert isinstance(exc, exc_type) + + +async def test_error_collection() -> None: + providers = {str(i): MockProvider() for i in range(6)} + subprovider0 = FallbackProvider({"0": providers["0"], "1": providers["1"]}, PriorityFallback()) + subprovider1 = FallbackProvider( + {"2": providers["2"], "3": providers["3"]}, PriorityFallback(), same_provider=True + ) + provider = FallbackProvider( + {"s0": subprovider0, "s1": subprovider1, "s2": providers["4"], "s3": providers["5"]}, + PriorityFallback(), ) - provider = FallbackProvider([subprovider1, subprovider2], PriorityFallback()) + async with provider.session() as session: - with pytest.raises(ValueError, match=r"Invalid provider path: \(2, 0\)"): - await session.rpc_at_pin((2, 0), random_request()) + providers["0"].set_state(ProviderState.UNREACHABLE) + _result, _path = await session.rpc_and_pin(random_request()) + assert_errors( + provider.errors(), + [ + (ProviderPath(("s0", "0")), Unreachable), + ], + ) + + # Now the whole `subprovider1` is unreachable + providers["1"].set_state(ProviderState.UNREACHABLE) + providers["2"].set_state(ProviderState.RPC_ERROR) + _result, _path = await session.rpc_and_pin(random_request()) + assert_errors( + provider.errors(), + [ + (ProviderPath(("s0", "0")), Unreachable), + (ProviderPath(("s0", "1")), Unreachable), + (ProviderPath(("s1", "2")), RPCError), + ], + ) + + # This will override the recorded error + _result, _path = await session.rpc_and_pin(random_request()) + providers["0"].set_state(ProviderState.BAD_RESPONSE) + providers["1"].set_state(ProviderState.NORMAL) + _result, _path = await session.rpc_and_pin(random_request()) + assert_errors( + provider.errors(), + [ + (ProviderPath(("s0", "0")), InvalidResponse), + (ProviderPath(("s0", "1")), Unreachable), + (ProviderPath(("s1", "2")), RPCError), + ], + ) + + # Test that the top fallback provider errors are included + providers["1"].set_state(ProviderState.UNREACHABLE) + providers["3"].set_state(ProviderState.UNREACHABLE) + providers["4"].set_state(ProviderState.PROTOCOL_ERROR) + _result, _path = await session.rpc_and_pin(random_request()) + assert_errors( + provider.errors(), + [ + (ProviderPath(("s2",)), ProtocolError), + (ProviderPath(("s0", "0")), InvalidResponse), + (ProviderPath(("s0", "1")), Unreachable), + (ProviderPath(("s1", "2")), RPCError), + (ProviderPath(("s1", "3")), Unreachable), + ], + ) diff --git a/tests/test_provider.py b/tests/test_provider.py index fcebadc..c83f6d5 100644 --- a/tests/test_provider.py +++ b/tests/test_provider.py @@ -22,7 +22,7 @@ Unreachable, _http_provider_server, # For monkeypatching purposes ) -from pons._provider import RPC_JSON, ProviderSession +from pons._provider import RPC_JSON, ProviderPath, ProviderSession @pytest.fixture @@ -42,6 +42,23 @@ async def session(test_server: HTTPProviderServer) -> AsyncIterator[ClientSessio yield session +def test_provider_path() -> None: + assert ProviderPath.empty().is_empty() + assert not ProviderPath(["1"]).is_empty() + + assert ProviderPath(["1", "2"]) == ProviderPath(["1", "2"]) + assert ProviderPath(["1", "2"]) != ProviderPath(["1", "3"]) + + assert hash(ProviderPath(["1", "2"])) == hash(ProviderPath(["1", "2"])) + assert hash(ProviderPath(["1", "2"])) != hash(ProviderPath(["1", "3"])) + + assert str(ProviderPath(["1", "2"])) == "1/2" + assert repr(ProviderPath(["1", "2"])) == "ProviderPath(('1', '2'))" + + assert ProviderPath(["1"]).group("2") == ProviderPath(["2", "1"]) + assert ProviderPath(["1", "2"]).ungroup() == ("1", ProviderPath(["2"])) + + async def test_single_value_request(session: ClientSession) -> None: assert await session.net_version() == "1" @@ -235,10 +252,10 @@ async def rpc(self, method: str, *_args: RPC_JSON) -> RPC_JSON: provider = MockProvider() async with provider.session() as session: result1 = await session.rpc_and_pin("1") - assert result1 == ("1", ()) + assert result1 == ("1", ProviderPath.empty()) - result2 = await session.rpc_at_pin((), "2") + result2 = await session.rpc_at_pin(ProviderPath.empty(), "2") assert result2 == "2" - with pytest.raises(ValueError, match=r"Unexpected provider path: \(1,\)"): - await session.rpc_at_pin((1,), "3") + with pytest.raises(ValueError, match=r"Expected an empty provider path, got: `1`"): + await session.rpc_at_pin(ProviderPath(["1"]), "3")