Implement retries on gRPC stream level

composer.json

@@ -37,7 +37,7 @@
         "thesis/googleapis-rpc-types": "^0.1.6",
         "thesis/package-version": "^0.1.2",
         "thesis/protobuf": "^0.1.8",
-        "thesis/protobuf-known-types": "^0.1.5"
+        "thesis/protobuf-known-types": "^0.1.7"
     },
     "require-dev": {
         "ext-bcmath": "*",
@@ -53,7 +53,12 @@
             "type": "path",
             "url": "packages/*",
             "options": {
-                "symlink": true
+                "symlink": true,
+                "versions": {
+                    "thesis/grpc-client": "0.1.x-dev",
+                    "thesis/grpc-protocol": "0.1.x-dev",
+                    "thesis/grpc-server": "0.1.x-dev"
+                }
             }
         }
     ],

packages/client/src/Client/Builder.php

@@ -9,6 +9,7 @@
 use Amp\Http\Client\DelegateHttpClient;
 use Amp\Http\Client\HttpClientBuilder;
 use Amp\Socket\ConnectContext;
+use Amp\Socket\DnsSocketConnector;
 use Amp\Socket\SocketConnector;
 use Thesis\Grpc\Client;
 use Thesis\Grpc\Client\Internal\Connection;
@@ -62,6 +63,8 @@ final class Builder
 
     private ?LoadBalancerFactory $loadBalancerFactory = null;
 
+    private ?Retry $retry = null;
+
     /** @var \SplObjectStorage<Scheme, EndpointResolver> */
     private \SplObjectStorage $endpointResolvers;
 
@@ -186,6 +189,14 @@ public function withLoadBalancer(LoadBalancerFactory $factory): self
         return $builder;
     }
 
+    public function withRetry(Retry $policy): self
+    {
+        $builder = clone $this;
+        $builder->retry = $policy;
+
+        return $builder;
+    }
+
     public function withEndpointResolver(Scheme $scheme, EndpointResolver $resolver): self
     {
         $builder = clone $this;
@@ -207,6 +218,7 @@ public function build(): Client
         $compressor = $this->compressor ?? IdentityCompressor::Compressor;
         $protobuf = $this->protobuf ?? Decoder\Builder::buildDefault();
         $loadBalancerFactory = $this->loadBalancerFactory ?? new LoadBalancer\PickFirstFactory();
+        $retry = $this->retry ?? Retry::disabled();
         $tlsContext = $this->credentials?->createContext();
         $uriFactory = new Http2\UriFactory($tlsContext !== null ? Internal\HttpScheme::Https : Internal\HttpScheme::Http);
         $transferTimeout = $this->transferTimeout;
@@ -230,7 +242,7 @@ public function build(): Client
             ->usingPool(ConnectionLimitingPool::byAuthority(
                 $this->connectionLimit,
                 new DefaultConnectionFactory(
-                    $this->connector,
+                    $this->connector ?? new DnsSocketConnector(),
                     new ConnectContext()
                         ->withConnectTimeout($this->connectTimeout)
                         ->withTlsContext($tlsContext),
@@ -257,6 +269,7 @@ public function build(): Client
                         encoder: $encoder,
                         compressor: $compressor,
                     ),
+                    retry: $retry,
                 ),
             ),
         );

packages/client/src/Client/Internal/AmphpHttpClient.php

@@ -26,16 +26,12 @@ public function invoke(
         Metadata $md = new Metadata(),
         Cancellation $cancellation = new NullCancellation(),
     ): object {
-        $stream = $this->connection->createStream(
+        return $this->connection->invoke(
+            $request,
             $invoke,
             $md,
             $cancellation,
         );
-
-        $stream->send($request);
-        $stream->close();
-
-        return $stream->receive();
     }
 
     #[\Override]

packages/client/src/Client/Internal/Connection.php

@@ -8,13 +8,29 @@
 use Amp\NullCancellation;
 use Thesis\Grpc\Client\Invoke;
 use Thesis\Grpc\ClientStream;
+use Thesis\Grpc\InvokeError;
 use Thesis\Grpc\Metadata;
 
 /**
  * @internal
  */
 interface Connection
 {
+    /**
+     * @template In of object
+     * @template Out of object
+     * @param In $request
+     * @param Invoke<In, Out> $invoke
+     * @return Out
+     * @throws InvokeError
+     */
+    public function invoke(
+        object $request,
+        Invoke $invoke,
+        Metadata $md = new Metadata(),
+        Cancellation $cancellation = new NullCancellation(),
+    ): object;
+
     /**
      * @template In of object
      * @template Out of object

packages/client/src/Client/Internal/Connection/DefaultConnection.php

@@ -7,6 +7,7 @@
 use Amp\Cancellation;
 use Amp\DeferredCancellation;
 use Amp\NullCancellation;
+use Thesis\Grpc\Client\Endpoint;
 use Thesis\Grpc\Client\EndpointResolver;
 use Thesis\Grpc\Client\EndpointResolverListener;
 use Thesis\Grpc\Client\Internal\Connection;
@@ -17,6 +18,7 @@
 use Thesis\Grpc\Client\LoadBalancerFactory;
 use Thesis\Grpc\Client\PickContext;
 use Thesis\Grpc\Client\Resolution;
+use Thesis\Grpc\Client\Retry;
 use Thesis\Grpc\Client\Target;
 use Thesis\Grpc\ClientStream;
 use Thesis\Grpc\Metadata;
@@ -38,6 +40,7 @@ public function __construct(
         LoadBalancerFactory $loadBalancerFactory,
         private InterceptorComposer $interceptor,
         private StreamFactory $streams,
+        private Retry $retry,
     ) {
         $this->deferredCancellation = new DeferredCancellation();
 
@@ -51,26 +54,47 @@ public function __construct(
     }
 
     #[\Override]
-    public function createStream(
+    public function invoke(
+        object $request,
         Invoke $invoke,
         Metadata $md = new Metadata(),
         Cancellation $cancellation = new NullCancellation(),
-    ): ClientStream {
-        $endpoint = $this->balancer->pick(new PickContext($invoke->method, $md));
+    ): object {
+        /** @var list<Endpoint> $excluded */
+        $excluded = [];
 
-        return $this->interceptor->intercept( // @phpstan-ignore return.type
-            $invoke,
-            $md,
-            $cancellation,
-            fn(Invoke $invoke, Metadata $md, Cancellation $cancellation) => $this->streams->create(
+        return $this->retry->call(
+            function () use (
+                &$excluded,
+                $request,
                 $invoke,
-                $endpoint->address,
                 $md,
                 $cancellation,
-            ),
+            ): object {
+                $endpoint = $this->balancer->pick(new PickContext($invoke->method, $md, $excluded));
+                $excluded[] = $endpoint;
+
+                $stream = $this->stream($endpoint, $invoke, $md, $cancellation);
+                $stream->send($request);
+                $stream->close();
+
+                return $stream->receive();
+            },
+            $cancellation,
         );
     }
 
+    #[\Override]
+    public function createStream(
+        Invoke $invoke,
+        Metadata $md = new Metadata(),
+        Cancellation $cancellation = new NullCancellation(),
+    ): ClientStream {
+        $endpoint = $this->balancer->pick(new PickContext($invoke->method, $md));
+
+        return $this->stream($endpoint, $invoke, $md, $cancellation);
+    }
+
     #[\Override]
     public function close(Cancellation $cancellation = new NullCancellation()): void
     {
@@ -84,4 +108,29 @@ public function onResolve(Resolution|\Throwable $result): void
             $this->balancer->refresh($result->endpoints);
         }
     }
+
+    /**
+     * @template In of object
+     * @template Out of object
+     * @param Invoke<In, Out> $invoke
+     * @return ClientStream<In, Out>
+     */
+    private function stream(
+        Endpoint $endpoint,
+        Invoke $invoke,
+        Metadata $md,
+        Cancellation $cancellation,
+    ): ClientStream {
+        return $this->interceptor->intercept( // @phpstan-ignore return.type
+            $invoke,
+            $md,
+            $cancellation,
+            fn(Invoke $invoke, Metadata $md, Cancellation $cancellation) => $this->streams->create(
+                $invoke,
+                $endpoint->address,
+                $md,
+                $cancellation,
+            ),
+        );
+    }
 }

packages/client/src/Client/Internal/Connection/LazyConnection.php

@@ -28,16 +28,27 @@ public function __construct(
         private readonly \Closure $factory,
     ) {}
 
+    #[\Override]
+    public function invoke(
+        object $request,
+        Invoke $invoke,
+        Metadata $md = new Metadata(),
+        Cancellation $cancellation = new NullCancellation(),
+    ): object {
+        return $this
+            ->createConnection($cancellation)
+            ->invoke($request, $invoke, $md, $cancellation);
+    }
+
     #[\Override]
     public function createStream(
         Invoke $invoke,
         Metadata $md = new Metadata(),
         Cancellation $cancellation = new NullCancellation(),
     ): ClientStream {
-        $this->future ??= async($this->factory);
-        $connection = $this->future->await($cancellation);
-
-        return $connection->createStream($invoke, $md, $cancellation);
+        return $this
+            ->createConnection($cancellation)
+            ->createStream($invoke, $md, $cancellation);
     }
 
     #[\Override]
@@ -48,4 +59,9 @@ public function close(Cancellation $cancellation = new NullCancellation()): void
 
         $future?->await($cancellation)->close($cancellation);
     }
+
+    private function createConnection(Cancellation $cancellation): Connection
+    {
+        return ($this->future ??= async($this->factory))->await($cancellation);
+    }
 }

packages/client/src/Client/Internal/Http2/ConcurrentClientStream.php

@@ -40,7 +40,10 @@ public function __construct(
         private readonly \Closure $decode,
         private readonly ErrorHandler $errors,
         private readonly Future $complete,
-    ) {}
+    ) {
+        $responseFuture->ignore();
+        $complete->ignore();
+    }
 
     #[\Override]
     public function send(object $message): void

packages/client/src/Client/LoadBalancer/PickFirst.php

@@ -20,7 +20,7 @@ final class PickFirst implements LoadBalancer
      * @param non-empty-list<Endpoint> $endpoints
      */
     public function __construct(
-        array $endpoints,
+        private array $endpoints,
         private readonly Randomizer $randomizer,
     ) {
         $this->current = $this->doPick($endpoints);
@@ -29,12 +29,20 @@ public function __construct(
     #[\Override]
     public function refresh(array $endpoints): void
     {
+        $this->endpoints = $endpoints;
         $this->current = $this->doPick($endpoints, $this->current);
     }
 
     #[\Override]
     public function pick(PickContext $context): Endpoint
     {
+        foreach ([$this->current, ...$this->endpoints] as $endpoint) {
+            if (!$context->excluded($endpoint)) {
+                $this->current = $endpoint;
+                break;
+            }
+        }
+
         return $this->current;
     }
 

packages/client/src/Client/LoadBalancer/RoundRobin.php

@@ -37,6 +37,14 @@ public function refresh(array $endpoints): void
     #[\Override]
     public function pick(PickContext $context): Endpoint
     {
+        for ($i = 0; $i < $this->count; ++$i) {
+            $endpoint = $this->endpoints[$this->cursor++ % $this->count]; // @phpstan-ignore offsetAccess.notFound
+
+            if (!$context->excluded($endpoint)) {
+                return $endpoint;
+            }
+        }
+
         return $this->endpoints[$this->cursor++ % $this->count]; // @phpstan-ignore offsetAccess.notFound
     }
 }

packages/client/src/Client/PickContext.php

@@ -13,9 +13,16 @@
 {
     /**
      * @param non-empty-string $methodName
+     * @param list<Endpoint> $excluded endpoints that already failed this call and should be skipped if possible
      */
     public function __construct(
         public string $methodName,
         public Metadata $metadata,
+        private array $excluded = [],
     ) {}
+
+    public function excluded(Endpoint $endpoint): bool
+    {
+        return array_any($this->excluded, $endpoint->equals(...));
+    }
 }