diff --git a/README.md b/README.md index d932cc1..fb75847 100644 --- a/README.md +++ b/README.md @@ -44,10 +44,10 @@ https://contractless.dev/contractless/Contractless-PHP-Modules ## Installation -Install version `0.2` through Packagist: +Install version `0.3` through Packagist: ```bash -composer require contractless/contractless-php-rpc:^0.2.0 -W +composer require contractless/contractless-php-rpc:^0.3.0 -W ``` Composer installs the package under `vendor/` and generates the required @@ -89,6 +89,79 @@ mainnet RPC unless the node operator configured another port. Every RPC call performs a signed client handshake. The wallet used for the handshake does not have to own an address being looked up. +## RPC Endpoint Pool + +Contractless client RPC is intentionally one request per connection. A node +accepts the client connection, performs the authenticated handshake, returns +one RPC reply, and closes the connection. + +`EndpointPoolTransport` is therefore a pool of available endpoints, not a +pool of persistent sockets. It randomly selects a new endpoint for every RPC +call so applications can distribute their requests across several nodes: + +```php +use Contractless\Rpc\Transport\EndpointPoolTransport; +use Contractless\Rpc\Transport\StreamTransport; + +$transport = new EndpointPoolTransport([ + 'local-node' => new StreamTransport( + host: '127.0.0.1', + port: 50050, + timeout: 10.0, + ), + 'public-node-1' => new StreamTransport( + host: 'node1.example.com', + port: 50050, + timeout: 10.0, + ), + 'public-node-2' => new StreamTransport( + host: 'node2.example.com', + port: 50050, + timeout: 10.0, + ), +]); + +$client = new Client( + $transport, + $crypto, + $credentials, +); +``` + +For every `Client` call, the pool: + +1. Randomizes the configured endpoint order. +2. Opens a one-shot RPC connection to the first selected endpoint. +3. Returns its complete reply when successful. +4. Tries the next randomized endpoint only after a `TransportException`. +5. Throws one `TransportException` identifying every attempted endpoint when + all transports fail. + +Protocol errors and node validation replies do not trigger failover. Those +responses may indicate that the request itself is invalid, so sending it to +additional nodes would add unnecessary traffic. + +The endpoint that most recently completed a request is available through: + +```php +$endpointId = $transport->lastSuccessfulEndpointId(); +``` + +Linked operations that must return to one endpoint can retrieve its configured +transport explicitly: + +```php +$selectedTransport = $transport->transportFor($endpointId); +$selectedClient = new Client( + $selectedTransport, + $crypto, + $credentials, +); +``` + +This is needed for operations such as paid storage lookup, where the fee quote, +payment address, and final lookup belong to the node that issued the quote. + ## RPC Replies RPC methods return the node's original reply as a PHP string. Depending on the diff --git a/src/Transport/EndpointPoolTransport.php b/src/Transport/EndpointPoolTransport.php new file mode 100644 index 0000000..3497bc7 --- /dev/null +++ b/src/Transport/EndpointPoolTransport.php @@ -0,0 +1,111 @@ + */ + private array $transports; + + private ?string $lastSuccessfulEndpointId = null; + + /** + * @param array $transports + */ + public function __construct(array $transports, private readonly bool $randomize = true) + { + if ($transports === []) { + throw new InvalidArgumentException('The RPC endpoint pool cannot be empty.'); + } + + foreach ($transports as $endpointId => $transport) { + if (!is_string($endpointId) || trim($endpointId) === '') { + throw new InvalidArgumentException( + 'Every RPC endpoint must have a non-empty string identifier.', + ); + } + if (!$transport instanceof TransportInterface) { + throw new InvalidArgumentException( + "RPC endpoint $endpointId does not implement TransportInterface.", + ); + } + } + + $this->transports = $transports; + } + + public function exchange( + string $handshake, + string $request, + string $uid, + Closure $validateHandshake, + ): string { + $this->lastSuccessfulEndpointId = null; + $endpointIds = array_keys($this->transports); + if ($this->randomize && count($endpointIds) > 1) { + shuffle($endpointIds); + } + + $failures = []; + $lastFailure = null; + + foreach ($endpointIds as $endpointId) { + try { + $reply = $this->transports[$endpointId]->exchange( + $handshake, + $request, + $uid, + $validateHandshake, + ); + $this->lastSuccessfulEndpointId = $endpointId; + return $reply; + } catch (TransportException $error) { + // Only transport failures move the same request to another + // endpoint. Protocol and validation errors remain final. + $failures[] = "$endpointId: {$error->getMessage()}"; + $lastFailure = $error; + } + } + + throw new TransportException( + 'Every Contractless RPC endpoint failed: ' . implode('; ', $failures), + previous: $lastFailure, + ); + } + + /** + * Return a specific transport for a linked request that must use the same + * endpoint as an earlier operation, such as a paid storage lookup. + */ + public function transportFor(string $endpointId): TransportInterface + { + if (!isset($this->transports[$endpointId])) { + throw new InvalidArgumentException("Unknown RPC endpoint: $endpointId"); + } + return $this->transports[$endpointId]; + } + + /** @return list */ + public function endpointIds(): array + { + return array_keys($this->transports); + } + + public function lastSuccessfulEndpointId(): ?string + { + return $this->lastSuccessfulEndpointId; + } +} diff --git a/tests/run.php b/tests/run.php index e1ca224..c5bb1be 100644 --- a/tests/run.php +++ b/tests/run.php @@ -17,8 +17,11 @@ spl_autoload_register(static function (string $class): void { use Contractless\Rpc\Client; use Contractless\Rpc\Crypto\CallbackCrypto; +use Contractless\Rpc\Exception\ProtocolException; +use Contractless\Rpc\Exception\TransportException; use Contractless\Rpc\Protocol\Address; use Contractless\Rpc\Protocol\Command; +use Contractless\Rpc\Transport\EndpointPoolTransport; use Contractless\Rpc\Transport\TransportInterface; use Contractless\Rpc\Transaction\TransferBuilder; use Contractless\Rpc\Transaction\AssetBuilder; @@ -36,6 +39,30 @@ function expect(bool $condition, string $message): void } } +final class TestTransport implements TransportInterface +{ + public int $calls = 0; + + public function __construct( + private readonly ?string $reply = null, + private readonly ?Throwable $failure = null, + ) { + } + + public function exchange( + string $handshake, + string $request, + string $uid, + Closure $validateHandshake, + ): string { + $this->calls++; + if ($this->failure !== null) { + throw $this->failure; + } + return $this->reply ?? ''; + } +} + expect( Address::toBytes(str_repeat('ab', 20) . '.cltc') === str_repeat("\xab", 20) . ".\x02", @@ -138,4 +165,91 @@ $loan = $agreements->loan('CLTC', 1000, $address, 'TEST', 2000, str_repeat('cd', $loan = $loan->signNext($crypto, $credentials); expect(strlen($loan->toSignedTransaction()->bytes) === 1492, 'Loan size is wrong.'); +$failedEndpoint = new TestTransport( + failure: new TransportException('Connection refused.'), +); +$workingEndpoint = new TestTransport(reply: 'pool reply'); +$endpointPool = new EndpointPoolTransport( + [ + 'offline-node' => $failedEndpoint, + 'working-node' => $workingEndpoint, + ], + randomize: false, +); +$pooledClient = new Client($endpointPool, $crypto, $credentials); +expect( + $pooledClient->blockHeight() === 'pool reply', + 'Endpoint pool did not return the successful fallback reply.', +); +expect($failedEndpoint->calls === 1, 'Endpoint pool did not try the first endpoint.'); +expect($workingEndpoint->calls === 1, 'Endpoint pool did not try the fallback endpoint.'); +expect( + $endpointPool->lastSuccessfulEndpointId() === 'working-node', + 'Endpoint pool did not record the endpoint that answered.', +); +expect( + $endpointPool->endpointIds() === ['offline-node', 'working-node'], + 'Endpoint pool did not preserve its endpoint identifiers.', +); +expect( + $endpointPool->transportFor('working-node') === $workingEndpoint, + 'Endpoint pool could not return a specifically requested transport.', +); + +$protocolFailure = new TestTransport( + failure: new ProtocolException('Invalid RPC reply.'), +); +$unusedEndpoint = new TestTransport(reply: 'must not run'); +$protocolPool = new EndpointPoolTransport( + [ + 'invalid-node' => $protocolFailure, + 'unused-node' => $unusedEndpoint, + ], + randomize: false, +); +$protocolClient = new Client($protocolPool, $crypto, $credentials); +$protocolRejected = false; +try { + $protocolClient->blockHeight(); +} catch (ProtocolException) { + $protocolRejected = true; +} +expect($protocolRejected, 'Endpoint pool swallowed a protocol error.'); +expect( + $unusedEndpoint->calls === 0, + 'Endpoint pool retried a request after a protocol error.', +); + +$allFailedPool = new EndpointPoolTransport( + [ + 'offline-one' => new TestTransport( + failure: new TransportException('First endpoint failed.'), + ), + 'offline-two' => new TestTransport( + failure: new TransportException('Second endpoint failed.'), + ), + ], + randomize: false, +); +$allFailedClient = new Client($allFailedPool, $crypto, $credentials); +$allFailedMessage = ''; +try { + $allFailedClient->blockHeight(); +} catch (TransportException $error) { + $allFailedMessage = $error->getMessage(); +} +expect( + str_contains($allFailedMessage, 'offline-one') + && str_contains($allFailedMessage, 'offline-two'), + 'Endpoint pool failure did not identify every failed endpoint.', +); + +$emptyPoolRejected = false; +try { + new EndpointPoolTransport([]); +} catch (InvalidArgumentException) { + $emptyPoolRejected = true; +} +expect($emptyPoolRejected, 'Endpoint pool accepted an empty endpoint list.'); + echo "All protocol tests passed.\n";