contractless
/
Contractless-PHP-RPC
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add randomized RPC endpoint pooling and failover

This commit is contained in:
viraladmin 2026-07-27 12:32:34 -06:00
parent d1f13d2acb
commit 3e60837ec3
2 changed files with 40 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
src/Transport/EndpointHealthTrackerInterface.php Normal file
View File

@ -0,0 +1,25 @@
<?php
declare(strict_types=1);
namespace Contractless\Rpc\Transport;
use Contractless\Rpc\Exception\TransportException;
interface EndpointHealthTrackerInterface
{
/**
* Remove endpoints that should not receive requests at this time.
*
* @param list<string> $endpointIds
* @return list<string>
*/
public function eligibleEndpointIds(array $endpointIds): array;
public function recordSuccess(string $endpointId): void;
public function recordFailure(
string $endpointId,
TransportException $error,
): void;
}

17
src/Transport/EndpointPoolTransport.php
View File

@ -25,8 +25,11 @@ final class EndpointPoolTransport implements TransportInterface
/**
* @param array<string, TransportInterface> $transports
*/
public function __construct(array $transports, private readonly bool $randomize = true)
{
public function __construct(
array $transports,
private readonly bool $randomize = true,
private readonly ?EndpointHealthTrackerInterface $healthTracker = null,
) {
if ($transports === []) {
throw new InvalidArgumentException('The RPC endpoint pool cannot be empty.');
}
@ -55,6 +58,14 @@ final class EndpointPoolTransport implements TransportInterface
): string {
$this->lastSuccessfulEndpointId = null;
$endpointIds = array_keys($this->transports);
if ($this->healthTracker !== null) {
$endpointIds = $this->healthTracker->eligibleEndpointIds($endpointIds);
}
if ($endpointIds === []) {
throw new TransportException(
'Every Contractless RPC endpoint is temporarily unavailable.',
);
}
if ($this->randomize && count($endpointIds) > 1) {
shuffle($endpointIds);
}
@ -71,10 +82,12 @@ final class EndpointPoolTransport implements TransportInterface
$validateHandshake,
);
$this->lastSuccessfulEndpointId = $endpointId;
$this->healthTracker?->recordSuccess($endpointId);
return $reply;
} catch (TransportException $error) {
// Only transport failures move the same request to another
// endpoint. Protocol and validation errors remain final.
$this->healthTracker?->recordFailure($endpointId, $error);
$failures[] = "$endpointId: {$error->getMessage()}";
$lastFailure = $error;
}