Skip to content

refactor(client): handle a failed status in the retry loop without try #849

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
xepozz wants to merge 1 commit into
base: master
Choose a base branch
from
Open

refactor(client): handle a failed status in the retry loop without try #849

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
109 changes: 54 additions & 55 deletions src/Client/GRPC/BaseClient.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -271,73 +271,72 @@ private function call(string $method, object $arg, ContextInterface $ctx): objec

do {
++$attempt;
try {
$options = $ctx->getOptions();
if ($deadline !== null) {
$diff = (new \DateTime())->diff($deadline);
$options['timeout'] = \max(0, (int) CarbonInterval::instance($diff)->totalMicroseconds);
}

/** @var UnaryCall $call */
$call = $this->connection->getClient()->{$method}($arg, $ctx->getMetadata(), $options);
[$result, $status] = $call->wait();
$options = $ctx->getOptions();
if ($deadline !== null) {
$diff = (new \DateTime())->diff($deadline);
$options['timeout'] = \max(0, (int) CarbonInterval::instance($diff)->totalMicroseconds);
}

if ($status->code !== 0) {
throw new ServiceClientException($status);
}
/** @var UnaryCall $call */
$call = $this->connection->getClient()->{$method}($arg, $ctx->getMetadata(), $options);
[$result, $status] = $call->wait();

if ($status->code === 0) {
\assert($result !== null);

return $result;
} catch (ServiceClientException $e) {
if (!\in_array($e->getCode(), self::RETRYABLE_ERRORS, true)) {
if ($e->getCode() === StatusCode::DEADLINE_EXCEEDED) {
throw new TimeoutException($e->getMessage(), $e->getCode(), $e);
}

if ($e->getCode() === StatusCode::CANCELLED) {
throw new CanceledException($e->getMessage(), $e->getCode(), $e);
}

// non retryable
throw $e;
}
}

$e = new ServiceClientException($status);

if ($retryOption->maximumAttempts !== 0 && $attempt >= $retryOption->maximumAttempts) {
// Reached maximum attempts
throw $e;
if (!\in_array($e->getCode(), self::RETRYABLE_ERRORS, true)) {
if ($e->getCode() === StatusCode::DEADLINE_EXCEEDED) {
throw new TimeoutException($e->getMessage(), $e->getCode(), $e);
}

// Init interval values in milliseconds
$initialIntervalMs ??= $retryOption->initialInterval === null
? (int) CarbonInterval::millisecond(50)->totalMilliseconds
: (int) (new CarbonInterval($retryOption->initialInterval))->totalMilliseconds;
$congestionInitialIntervalMs ??= $retryOption->congestionInitialInterval === null
? (int) CarbonInterval::millisecond(1000)->totalMilliseconds
: (int) (new CarbonInterval($retryOption->congestionInitialInterval))->totalMilliseconds;

$throttler ??= new BackoffThrottler(
maxInterval: $retryOption->maximumInterval !== null
? (int) (new CarbonInterval($retryOption->maximumInterval))->totalMilliseconds
: $initialIntervalMs * 200,
maxJitterCoefficient: $retryOption->maximumJitterCoefficient,
backoffCoefficient: $retryOption->backoffCoefficient,
);

// Initial interval always depends on the *most recent* failure.
$baseInterval = $e->getCode() === StatusCode::RESOURCE_EXHAUSTED
? $congestionInitialIntervalMs
: $initialIntervalMs;

$wait = $throttler->calculateSleepTime(failureCount: $attempt, initialInterval: $baseInterval) * 1000;

if ($deadline !== null && new \DateTimeImmutable("+{$wait} microseconds") > $deadline) {
throw new TimeoutException('Call timeout has been reached', $e->getCode(), $e);
if ($e->getCode() === StatusCode::CANCELLED) {
throw new CanceledException($e->getMessage(), $e->getCode(), $e);
}

// wait till the next call
$this->usleep($wait);
// non retryable
throw $e;
}

if ($retryOption->maximumAttempts !== 0 && $attempt >= $retryOption->maximumAttempts) {
// Reached maximum attempts
throw $e;
}

// Init interval values in milliseconds
$initialIntervalMs ??= $retryOption->initialInterval === null
? (int) CarbonInterval::millisecond(50)->totalMilliseconds
: (int) (new CarbonInterval($retryOption->initialInterval))->totalMilliseconds;
$congestionInitialIntervalMs ??= $retryOption->congestionInitialInterval === null
? (int) CarbonInterval::millisecond(1000)->totalMilliseconds
: (int) (new CarbonInterval($retryOption->congestionInitialInterval))->totalMilliseconds;

$throttler ??= new BackoffThrottler(
maxInterval: $retryOption->maximumInterval !== null
? (int) (new CarbonInterval($retryOption->maximumInterval))->totalMilliseconds
: $initialIntervalMs * 200,
maxJitterCoefficient: $retryOption->maximumJitterCoefficient,
backoffCoefficient: $retryOption->backoffCoefficient,
);

// Initial interval always depends on the *most recent* failure.
$baseInterval = $e->getCode() === StatusCode::RESOURCE_EXHAUSTED
? $congestionInitialIntervalMs
: $initialIntervalMs;

$wait = $throttler->calculateSleepTime(failureCount: $attempt, initialInterval: $baseInterval) * 1000;

if ($deadline !== null && new \DateTimeImmutable("+{$wait} microseconds") > $deadline) {
throw new TimeoutException('Call timeout has been reached', $e->getCode(), $e);
}

// wait till the next call
$this->usleep($wait);
} while (true);
}

Expand Down
30 changes: 9 additions & 21 deletions tests/Unit/Client/GRPC/BaseClientTestCase.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,6 @@
use Temporal\Client\GRPC\Connection\ConnectionState;
use Temporal\Client\GRPC\ContextInterface;
use Temporal\Client\GRPC\ServiceClient;
use Temporal\Client\GRPC\StatusCode;
use Temporal\Exception\Client\ServiceClientException;
use Temporal\Exception\Client\TimeoutException;
use Temporal\Internal\Interceptor\Pipeline;
Expand Down Expand Up @@ -163,10 +162,10 @@ public function __construct(
private readonly \ArrayObject $timeouts,
) {}

public function testCall(object $arg, array $metadata, array $options): void
public function testCall(object $arg, array $metadata, array $options): object
{
$this->timeouts[] = $options['timeout'];
throw new class((object) ['code' => StatusCode::UNKNOWN, 'metadata' => []]) extends ServiceClientException {};
return new FailedCall();
}

public function close(): void {}
Expand Down Expand Up @@ -195,10 +194,10 @@ public function __construct(
private readonly \ArrayObject $timeouts,
) {}

public function testCall(object $arg, array $metadata, array $options): void
public function testCall(object $arg, array $metadata, array $options): object
{
$this->timeouts[] = $options['timeout'];
throw new class((object) ['code' => StatusCode::UNKNOWN, 'metadata' => []]) extends ServiceClientException {};
return new FailedCall();
}

public function close(): void {}
Expand Down Expand Up @@ -230,9 +229,9 @@ public function testServiceClientCallTimeoutDoesNotSleepPastTheDeadline(): void
$client = $this->createClientMock(static fn() => new class extends WorkflowServiceClient {
public function __construct() {}

public function testCall(): void
public function testCall(): object
{
throw new class((object) ['code' => StatusCode::UNKNOWN, 'metadata' => []]) extends ServiceClientException {};
return new FailedCall();
}

public function close(): void {}
Expand Down Expand Up @@ -292,20 +291,10 @@ public function testServiceClientCallMaximumAttemptsReached(): void
$client = $this->createClientMock(fn() => new class extends WorkflowServiceClient {
public function __construct() {}

public function testCall(): void
public function testCall(): object
{
static $counter = 0;
throw new class(++$counter) extends ServiceClientException {
public function __construct(public int $attempt)
{
parent::__construct((object) ['code' => StatusCode::UNKNOWN, 'metadata' => []]);
}

public function isTestError(): bool
{
return true;
}
};
return new FailedCall('attempt ' . ++$counter);
}

public function close(): void {}
Expand All @@ -321,8 +310,7 @@ public function close(): void {}
$client->testCall();
self::fail('Expected exception');
} catch (ServiceClientException $e) {
self::assertTrue($e->isTestError());
self::assertSame(3, $e->attempt);
self::assertSame('attempt 3 (code: 2)', $e->getMessage());
}
}

Expand Down
19 changes: 19 additions & 0 deletions tests/Unit/Client/GRPC/FailedCall.php
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,19 @@
<?php

declare(strict_types=1);

namespace Temporal\Tests\Unit\Client\GRPC;

use Temporal\Client\GRPC\StatusCode;

final class FailedCall
{
public function __construct(
private readonly string $details = '',
) {}

public function wait(): array
{
return [null, (object) ['code' => StatusCode::UNKNOWN, 'details' => $this->details, 'metadata' => []]];
}
}
Loading