[Client] Implement setMaxRetries connection retries (#413)
* [Client] Implement setMaxRetries connection retries * [Client] Reject a negative retry count in Configuration * [Client] Use a positive init timeout in the retry test * [Client] Drop redundant docblock note on setMaxRetries * [Client] Clear the initialized flag on a failed connection * [Client] Trim comments to the non-obvious ones * [Client] Cover retries and timeout cleanup with integration tests The retry fixture counts the processes it was started as, which is what separates a real retry from a second call into the same server. The timeout fixture outlives the client's patience and then goes idle, so the next request runs on a connection the client already gave up on once.
This commit is contained in:
parent
7b12dc4010
commit
d8cc21a345
4 changed files with 616 additions and 26 deletions
|
|
@ -58,6 +58,8 @@ use Psr\Log\NullLogger;
|
|||
*/
|
||||
class Client
|
||||
{
|
||||
private const RETRY_BASE_DELAY_MS = 100;
|
||||
|
||||
private ?TransportInterface $transport = null;
|
||||
|
||||
public function __construct(
|
||||
|
|
@ -78,16 +80,44 @@ class Client
|
|||
/**
|
||||
* Connect to an MCP server using the provided transport.
|
||||
*
|
||||
* @throws ConnectionException If connection or initialization fails
|
||||
* A failed attempt is closed and retried, see {@see Builder::setMaxRetries()}.
|
||||
*
|
||||
* @throws ConnectionException If connection or initialization fails on every attempt
|
||||
*/
|
||||
public function connect(TransportInterface $transport): void
|
||||
{
|
||||
$this->transport = $transport;
|
||||
$this->protocol->connect($transport, $this->config);
|
||||
|
||||
$transport->connect();
|
||||
$maxAttempts = $this->config->maxRetries + 1;
|
||||
|
||||
$this->logger->info('Client connected and initialized');
|
||||
for ($attempt = 1; $attempt <= $maxAttempts; ++$attempt) {
|
||||
try {
|
||||
$transport->connect();
|
||||
|
||||
$this->logger->info('Client connected and initialized', ['attempt' => $attempt]);
|
||||
|
||||
return;
|
||||
} catch (ConnectionException $e) {
|
||||
// initialize() flags the session before sending the initialized
|
||||
// notification, so a failure in between leaves the flag set.
|
||||
$this->protocol->getState()->setInitialized(false);
|
||||
|
||||
$transport->close();
|
||||
|
||||
if ($attempt === $maxAttempts) {
|
||||
throw $e;
|
||||
}
|
||||
|
||||
$this->logger->warning('Connection attempt failed, retrying', [
|
||||
'attempt' => $attempt,
|
||||
'max_attempts' => $maxAttempts,
|
||||
'exception' => $e,
|
||||
]);
|
||||
|
||||
usleep($attempt * self::RETRY_BASE_DELAY_MS * 1000);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -96,7 +96,11 @@ final class Builder
|
|||
}
|
||||
|
||||
/**
|
||||
* Set maximum retry attempts for failed connections.
|
||||
* Set the number of times a failed connection attempt is retried.
|
||||
*
|
||||
* Counts retries, not attempts: 3 means one initial attempt plus up to three
|
||||
* retries. Pass 0 to fail on the first failure. Only applies to
|
||||
* {@see Client::connect()}; individual requests are never retried.
|
||||
*/
|
||||
public function setMaxRetries(int $retries): self
|
||||
{
|
||||
|
|
|
|||
|
|
@ -38,5 +38,9 @@ class Configuration
|
|||
if ($requestTimeout < 1) {
|
||||
throw new InvalidArgumentException(\sprintf('The request timeout must be a positive number of seconds, got %d.', $requestTimeout));
|
||||
}
|
||||
|
||||
if ($maxRetries < 0) {
|
||||
throw new InvalidArgumentException(\sprintf('The maximum number of retries must be zero or greater, got %d.', $maxRetries));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -179,22 +179,29 @@ class Protocol
|
|||
}
|
||||
|
||||
$this->state->addPendingRequest($requestId, $timeout);
|
||||
$this->sendRequest($request);
|
||||
|
||||
$immediate = $this->state->consumeResponse($requestId);
|
||||
if (null !== $immediate) {
|
||||
$this->logger->debug('Received immediate response', ['id' => $requestId]);
|
||||
try {
|
||||
$this->sendRequest($request);
|
||||
|
||||
return $immediate;
|
||||
$immediate = $this->state->consumeResponse($requestId);
|
||||
if (null !== $immediate) {
|
||||
$this->logger->debug('Received immediate response', ['id' => $requestId]);
|
||||
|
||||
return $immediate;
|
||||
}
|
||||
|
||||
$this->logger->debug('Suspending fiber for response', ['id' => $requestId]);
|
||||
|
||||
return \Fiber::suspend([
|
||||
'type' => 'await_response',
|
||||
'request_id' => $requestId,
|
||||
'timeout' => $timeout,
|
||||
]);
|
||||
} finally {
|
||||
// Only the response path clears it, so a request that timed out or
|
||||
// whose send() threw would stay pending and fail every later one.
|
||||
$this->state->removePendingRequest($requestId);
|
||||
}
|
||||
|
||||
$this->logger->debug('Suspending fiber for response', ['id' => $requestId]);
|
||||
|
||||
return \Fiber::suspend([
|
||||
'type' => 'await_response',
|
||||
'request_id' => $requestId,
|
||||
'timeout' => $timeout,
|
||||
]);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
Loading…
Reference in a new issue