gRPC interceptors refactoring

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/README.md

@@ -236,23 +236,43 @@ $client = new Client\Builder()
 
 ## Interceptors
 
-Interceptors let you add cross-cutting logic (auth, logging, tracing, retry, metadata enrichment) without changing service stubs.
+Interceptors let you add cross-cutting logic (auth, logging, tracing, metadata enrichment) without changing service stubs. Mirroring gRPC, they come in two flavours, registered on separate chains:
+
+- `UnaryInterceptor` wraps a whole unary `request → response` call (via `withUnaryInterceptors()`). A retry policy, for instance, is naturally a unary interceptor — re-invoking `$invoker` reuses the call's pick context, so the transport fails over to another endpoint automatically.
+- `StreamInterceptor` wraps the creation of a stream (via `withStreamInterceptors()`), and so also covers the underlying stream of every streaming RPC.
+
+A unary call does **not** run the stream chain and vice versa. An interceptor that must apply to both (e.g. auth) implements both interfaces and is registered on both chains.
 
 ```php
 use Amp\Cancellation;
 use Thesis\Grpc\Client;
 use Thesis\Grpc\Client\Invoke;
+use Thesis\Grpc\Client\StreamInterceptor;
+use Thesis\Grpc\Client\UnaryInterceptor;
 use Thesis\Grpc\ClientStream;
 use Thesis\Grpc\Metadata;
 
-final readonly class ClientAuthInterceptor implements Client\Interceptor
+final readonly class ClientAuthInterceptor implements UnaryInterceptor, StreamInterceptor
 {
     #[\Override]
-    public function intercept(Invoke $invoke, Metadata $md, Cancellation $cancellation, callable $next): ClientStream
+    public function interceptUnary(object $request, Invoke $invoke, Metadata $md, Cancellation $cancellation, callable $invoker): object
+    {
+        return $invoker($request, $invoke, $md->with('Authorization', 'supertoken'), $cancellation);
+    }
+
+    #[\Override]
+    public function interceptStream(Invoke $invoke, Metadata $md, Cancellation $cancellation, callable $newStream): ClientStream
     {
-        return $next($invoke, $md->with('Authorization', 'supertoken'), $cancellation);
+        return $newStream($invoke, $md->with('Authorization', 'supertoken'), $cancellation);
     }
 }
+
+$auth = new ClientAuthInterceptor();
+
+$client = new Client\Builder()
+    ->withUnaryInterceptors($auth)
+    ->withStreamInterceptors($auth)
+    ->build();
 ```
 
 ## Client streaming

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;
@@ -37,8 +38,11 @@ final class Builder
 
     private ?DelegateHttpClient $httpclient = null;
 
-    /** @var list<Interceptor> */
-    private array $interceptors = [];
+    /** @var list<UnaryInterceptor> */
+    private array $unaryInterceptors = [];
+
+    /** @var list<StreamInterceptor> */
+    private array $streamInterceptors = [];
 
     private ?TransportCredentials $credentials = null;
 
@@ -116,11 +120,25 @@ public function withHost(string $host): self
     /**
      * @no-named-arguments
      */
-    public function withInterceptors(Interceptor ...$interceptors): self
+    public function withUnaryInterceptors(UnaryInterceptor ...$interceptors): self
     {
         $builder = clone $this;
-        $builder->interceptors = [
-            ...$builder->interceptors,
+        $builder->unaryInterceptors = [
+            ...$builder->unaryInterceptors,
+            ...$interceptors,
+        ];
+
+        return $builder;
+    }
+
+    /**
+     * @no-named-arguments
+     */
+    public function withStreamInterceptors(StreamInterceptor ...$interceptors): self
+    {
+        $builder = clone $this;
+        $builder->streamInterceptors = [
+            ...$builder->streamInterceptors,
             ...$interceptors,
         ];
 
@@ -218,19 +236,28 @@ public function build(): Client
             Scheme::Ipv4, Scheme::Ipv6, Scheme::Unix => new EndpointResolver\StaticResolver(),
         };
 
-        $interceptor = new Http2\InterceptorComposer([
-            ...$this->interceptors,
-            new Http2\AppendControlMetadataInterceptor(
-                $encoder->name(),
-                $compressor->name(),
-            ),
+        $controlMetadata = new Internal\AppendControlMetadataInterceptor(
+            $encoder->name(),
+            $compressor->name(),
+        );
+
+        // Control metadata sits innermost (closest to the transport) so every user
+        // interceptor runs before the HTTP/2 headers are finalised.
+        $unary = new Internal\UnaryInterceptorComposer([
+            ...$this->unaryInterceptors,
+            $controlMetadata,
+        ]);
+
+        $stream = new Internal\StreamInterceptorComposer([
+            ...$this->streamInterceptors,
+            $controlMetadata,
         ]);
 
         $httpclient = $this->httpclient ?? new HttpClientBuilder()
             ->usingPool(ConnectionLimitingPool::byAuthority(
                 $this->connectionLimit,
                 new DefaultConnectionFactory(
-                    $this->connector,
+                    $this->connector ?? new DnsSocketConnector(),
                     new ConnectContext()
                         ->withConnectTimeout($this->connectTimeout)
                         ->withTlsContext($tlsContext),
@@ -247,7 +274,6 @@ public function build(): Client
                     target: $target,
                     resolver: $resolver,
                     loadBalancerFactory: $loadBalancerFactory,
-                    interceptor: $interceptor,
                     streams: new Http2\StreamFactory(
                         http: $httpclient,
                         uri: $uriFactory,
@@ -259,6 +285,8 @@ public function build(): Client
                     ),
                 ),
             ),
+            $unary,
+            $stream,
         );
     }
 }

packages/client/src/Client/CallableInterceptor.php → packages/client/src/Client/CallableStreamInterceptor.php

@@ -11,7 +11,7 @@
 /**
  * @api
  */
-final readonly class CallableInterceptor implements Interceptor
+final readonly class CallableStreamInterceptor implements StreamInterceptor
 {
     /**
      * @template In of object
@@ -23,17 +23,17 @@ public function __construct(
     ) {}
 
     #[\Override]
-    public function intercept(
+    public function interceptStream(
         Invoke $invoke,
         Metadata $md,
         Cancellation $cancellation,
-        callable $next,
+        callable $newStream,
     ): ClientStream {
         return ($this->handler)(
             $invoke,
             $md,
             $cancellation,
-            $next,
+            $newStream,
         );
     }
 }

packages/client/src/Client/CallableUnaryInterceptor.php

@@ -0,0 +1,40 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Grpc\Client;
+
+use Amp\Cancellation;
+use Thesis\Grpc\Metadata;
+
+/**
+ * @api
+ */
+final readonly class CallableUnaryInterceptor implements UnaryInterceptor
+{
+    /**
+     * @template In of object
+     * @template Out of object
+     * @param callable(In, Invoke<In, Out>, Metadata, Cancellation, callable(In, Invoke<In, Out>, Metadata, Cancellation): Out): Out $handler
+     */
+    public function __construct(
+        private mixed $handler,
+    ) {}
+
+    #[\Override]
+    public function interceptUnary(
+        object $request,
+        Invoke $invoke,
+        Metadata $md,
+        Cancellation $cancellation,
+        callable $invoker,
+    ): object {
+        return ($this->handler)(
+            $request,
+            $invoke,
+            $md,
+            $cancellation,
+            $invoker,
+        );
+    }
+}

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

@@ -6,8 +6,12 @@
 
 use Amp\Cancellation;
 use Amp\NullCancellation;
+use Google\Rpc\Code;
 use Thesis\Grpc\Client;
+use Thesis\Grpc\Client\PickContext;
 use Thesis\Grpc\ClientStream;
+use Thesis\Grpc\GrpcException;
+use Thesis\Grpc\InvokeError;
 use Thesis\Grpc\Metadata;
 
 /**
@@ -17,6 +21,8 @@
 {
     public function __construct(
         private Connection $connection,
+        private UnaryInterceptorComposer $unary,
+        private StreamInterceptorComposer $stream,
     ) {}
 
     #[\Override]
@@ -26,16 +32,34 @@ public function invoke(
         Metadata $md = new Metadata(),
         Cancellation $cancellation = new NullCancellation(),
     ): object {
-        $stream = $this->connection->createStream(
+        $pick = new PickContext($invoke->method, $md);
+
+        return $this->unary->intercept( // @phpstan-ignore return.type
+            $request,
             $invoke,
             $md,
             $cancellation,
-        );
-
-        $stream->send($request);
-        $stream->close();
+            function (
+                object $request,
+                Client\Invoke $invoke,
+                Metadata $md,
+                Cancellation $cancellation,
+            ) use ($pick): object {
+                try {
+                    $stream = $this->connection->createStream($invoke, $md, $cancellation, $pick);
+                    $stream->send($request);
+                    $stream->close();
 
-        return $stream->receive();
+                    return $stream->receive();
+                } catch (GrpcException $e) {
+                    throw $e;
+                } catch (\Throwable $e) {
+                    // Transport-level failures (e.g. a refused connection) map to UNAVAILABLE,
+                    // so interceptors above see a gRPC status rather than a raw amphp exception.
+                    throw new InvokeError(Code::UNAVAILABLE, $e->getMessage(), previous: $e);
+                }
+            },
+        );
     }
 
     #[\Override]
@@ -44,10 +68,17 @@ public function createStream(
         Metadata $md = new Metadata(),
         Cancellation $cancellation = new NullCancellation(),
     ): ClientStream {
-        return $this->connection->createStream(
+        $pick = new PickContext($invoke->method, $md);
+
+        return $this->stream->intercept( // @phpstan-ignore return.type
             $invoke,
             $md,
             $cancellation,
+            fn(
+                Client\Invoke $invoke,
+                Metadata $md,
+                Cancellation $cancellation,
+            ): ClientStream => $this->connection->createStream($invoke, $md, $cancellation, $pick),
         );
     }
 

packages/client/src/Client/Internal/Http2/AppendControlMetadataInterceptor.php → packages/client/src/Client/Internal/AppendControlMetadataInterceptor.php

@@ -2,18 +2,21 @@
 
 declare(strict_types=1);
 
-namespace Thesis\Grpc\Client\Internal\Http2;
+namespace Thesis\Grpc\Client\Internal;
 
 use Amp\Cancellation;
-use Thesis\Grpc\Client\Interceptor;
 use Thesis\Grpc\Client\Invoke;
+use Thesis\Grpc\Client\StreamInterceptor;
+use Thesis\Grpc\Client\UnaryInterceptor;
 use Thesis\Grpc\ClientStream;
 use Thesis\Grpc\Metadata;
 
 /**
  * @internal
  */
-final readonly class AppendControlMetadataInterceptor implements Interceptor
+final readonly class AppendControlMetadataInterceptor implements
+    UnaryInterceptor,
+    StreamInterceptor
 {
     /**
      * @param non-empty-string $encoding
@@ -25,18 +28,32 @@ public function __construct(
     ) {}
 
     #[\Override]
-    public function intercept(
+    public function interceptUnary(
+        object $request,
         Invoke $invoke,
         Metadata $md,
         Cancellation $cancellation,
-        callable $next,
+        callable $invoker,
+    ): object {
+        return $invoker($request, $invoke, $this->decorate($md), $cancellation);
+    }
+
+    #[\Override]
+    public function interceptStream(
+        Invoke $invoke,
+        Metadata $md,
+        Cancellation $cancellation,
+        callable $newStream,
     ): ClientStream {
-        $md = $md
+        return $newStream($invoke, $this->decorate($md), $cancellation);
+    }
+
+    private function decorate(Metadata $md): Metadata
+    {
+        return $md
             ->withKey(new Metadata\ContentType($this->encoding))
             ->withKey(Metadata\UserAgent::Key)
             ->withKey(new Metadata\ContentEncoding($this->compression))
             ->with('TE', 'trailers');
-
-        return $next($invoke, $md, $cancellation);
     }
 }