From df5b64b82502e5fc5accbe9d07354e57f9cd6c64 Mon Sep 17 00:00:00 2001 From: Rauan Mayemir Date: Mon, 9 Sep 2024 18:29:26 +0500 Subject: [PATCH 1/7] Improve handling grpc response headers, propagate response headers with grpc error metadata --- composer.json | 5 +++ src/Internal/CallContext.php | 40 ++++++++++++++++++++++ src/Server.php | 66 ++++++++++++++---------------------- 3 files changed, 70 insertions(+), 41 deletions(-) create mode 100644 src/Internal/CallContext.php diff --git a/composer.json b/composer.json index 49f9ef9..41325d1 100644 --- a/composer.json +++ b/composer.json @@ -77,6 +77,11 @@ "config": { "sort-packages": true }, + "extra": { + "branch-alias": { + "3.x": "3.4.x-dev" + } + }, "minimum-stability": "dev", "prefer-stable": true } diff --git a/src/Internal/CallContext.php b/src/Internal/CallContext.php new file mode 100644 index 0000000..0db223b --- /dev/null +++ b/src/Internal/CallContext.php @@ -0,0 +1,40 @@ + $service + * @param non-empty-string $method + * @param array> $context + */ + public function __construct( + public string $service, + public string $method, + public array $context, + ) { + } + + /** + * @throws \JsonException + */ + public static function decode(string $payload): self + { + $data = Json::decode($payload); + + return new self( + service: $data['service'], + method: $data['method'], + context: $data['context'], + ); + } +} diff --git a/src/Server.php b/src/Server.php index 7868569..a8cc019 100644 --- a/src/Server.php +++ b/src/Server.php @@ -10,6 +10,7 @@ use Spiral\RoadRunner\GRPC\Exception\GRPCExceptionInterface; use Spiral\RoadRunner\GRPC\Exception\NotFoundException; use Spiral\RoadRunner\GRPC\Exception\ServiceException; +use Spiral\RoadRunner\GRPC\Internal\CallContext; use Spiral\RoadRunner\GRPC\Internal\Json; use Spiral\RoadRunner\Payload; use Spiral\RoadRunner\Worker; @@ -21,12 +22,6 @@ * @psalm-type ServerOptions = array{ * debug?: bool * } - * - * @psalm-type ContextResponse = array{ - * service: class-string, - * method: non-empty-string, - * context: array> - * } */ final class Server { @@ -77,15 +72,30 @@ public function serve(?WorkerInterface $worker = null, ?callable $finalize = nul return; } + $responseHeaders = new ResponseHeaders(); + try { - /** @var ContextResponse $context */ - $context = Json::decode($request->header); + $call = CallContext::decode($request->header); + + $context = (new Context($call->context)) + ->withValue(ResponseHeaders::class, $responseHeaders); - [$answerBody, $answerHeaders] = $this->tick($request->body, $context); + $response = $this->invoke($call->service, $call->method, $context, $request->body); - $this->workerSend($worker, $answerBody, $answerHeaders); + $this->workerSend( + worker: $worker, + body: $response, + headers: $responseHeaders->packHeaders() + ); } catch (GRPCExceptionInterface $e) { - $this->workerGrpcError($worker, $e); + $this->workerSend( + worker: $worker, + body: '', + headers: Json::encode([ + 'error' => $this->createGrpcError($e), + 'headers' => $responseHeaders->packHeaders(), + ]), + ); } catch (\Throwable $e) { $this->workerError($worker, $this->isDebugMode() ? (string) $e : $e->getMessage()); } finally { @@ -112,24 +122,9 @@ protected function invoke(string $service, string $method, ContextInterface $con return $this->services[$service]->invoke($method, $context, $body); } - /** - * @param ContextResponse $data - * @return array{0: string, 1: string} - * @throws \JsonException - * @throws \Throwable - */ - private function tick(string $body, array $data): array + private function workerError(WorkerInterface $worker, string $message): void { - $context = (new Context($data['context'])) - ->withValue(ResponseHeaders::class, new ResponseHeaders()); - - $response = $this->invoke($data['service'], $data['method'], $context, $body); - - /** @var ResponseHeaders|null $responseHeaders */ - $responseHeaders = $context->getValue(ResponseHeaders::class); - $responseHeadersString = $responseHeaders ? $responseHeaders->packHeaders() : '{}'; - - return [$response, $responseHeadersString]; + $worker->error($message); } /** @@ -140,12 +135,7 @@ private function workerSend(WorkerInterface $worker, string $body, string $heade $worker->respond(new Payload($body, $headers)); } - private function workerError(WorkerInterface $worker, string $message): void - { - $worker->error($message); - } - - private function workerGrpcError(WorkerInterface $worker, GRPCExceptionInterface $e): void + private function createGrpcError(GRPCExceptionInterface $e): string { $status = new Status([ 'code' => $e->getCode(), @@ -161,13 +151,7 @@ static function ($detail) { ), ]); - $this->workerSend( - $worker, - '', - Json::encode([ - 'error' => \base64_encode($status->serializeToString()), - ]), - ); + return \base64_encode($status->serializeToString()); } /** From cfbad5adae6ccd4ce972e59a66bd7bccf30eb158 Mon Sep 17 00:00:00 2001 From: Rauan Mayemir Date: Tue, 10 Sep 2024 14:19:52 +0500 Subject: [PATCH 2/7] Add response trailers --- src/ResponseTrailers.php | 73 ++++++++++++++++++++++++++++++++++++++++ src/Server.php | 16 +++++++-- 2 files changed, 86 insertions(+), 3 deletions(-) create mode 100644 src/ResponseTrailers.php diff --git a/src/ResponseTrailers.php b/src/ResponseTrailers.php new file mode 100644 index 0000000..3231013 --- /dev/null +++ b/src/ResponseTrailers.php @@ -0,0 +1,73 @@ + + */ +final class ResponseTrailers implements \IteratorAggregate, \Countable +{ + /** + * @var array + */ + private array $trailers = []; + + /** + * @param iterable $trailers + */ + public function __construct(iterable $trailers = []) + { + foreach ($trailers as $key => $value) { + $this->set($key, $value); + } + } + + /** + * @param THeaderKey $key + * @param THeaderValue $value + */ + public function set(string $key, string $value): void + { + $this->trailers[$key] = $value; + } + + /** + * @param THeaderKey $key + * @param string|null $default + * @return THeaderValue|null + */ + public function get(string $key, string $default = null): ?string + { + return $this->trailers[$key] ?? $default; + } + + public function getIterator(): \Traversable + { + return new \ArrayIterator($this->trailers); + } + + public function count(): int + { + return \count($this->trailers); + } + + /** + * @throws \JsonException + */ + public function packTrailers(): string + { + // If an empty array is serialized, it is cast to the string "[]" + // instead of object string "{}" + if ($this->trailers === []) { + return '{}'; + } + + return Json::encode($this->trailers); + } +} diff --git a/src/Server.php b/src/Server.php index a8cc019..8331cf4 100644 --- a/src/Server.php +++ b/src/Server.php @@ -73,19 +73,28 @@ public function serve(?WorkerInterface $worker = null, ?callable $finalize = nul } $responseHeaders = new ResponseHeaders(); + $responseTrailers = new ResponseTrailers(); try { $call = CallContext::decode($request->header); - $context = (new Context($call->context)) - ->withValue(ResponseHeaders::class, $responseHeaders); + $context = new Context(array_merge( + $call->context, + [ + ResponseHeaders::class => $responseHeaders, + ResponseTrailers::class => $responseTrailers + ] + )); $response = $this->invoke($call->service, $call->method, $context, $request->body); $this->workerSend( worker: $worker, body: $response, - headers: $responseHeaders->packHeaders() + headers: Json::encode([ + 'headers' => $responseHeaders->packHeaders(), + 'trailers' => $responseTrailers->packTrailers(), + ]), ); } catch (GRPCExceptionInterface $e) { $this->workerSend( @@ -94,6 +103,7 @@ public function serve(?WorkerInterface $worker = null, ?callable $finalize = nul headers: Json::encode([ 'error' => $this->createGrpcError($e), 'headers' => $responseHeaders->packHeaders(), + 'trailers' => $responseTrailers->packTrailers(), ]), ); } catch (\Throwable $e) { From c593a031307d5a52b38285d1c826f12d731377ef Mon Sep 17 00:00:00 2001 From: Rauan Mayemir Date: Tue, 10 Sep 2024 15:12:43 +0500 Subject: [PATCH 3/7] Do not add empty headers/trailers --- src/Server.php | 22 +++++++++++++--------- 1 file changed, 13 insertions(+), 9 deletions(-) diff --git a/src/Server.php b/src/Server.php index 8331cf4..0d883be 100644 --- a/src/Server.php +++ b/src/Server.php @@ -88,23 +88,27 @@ public function serve(?WorkerInterface $worker = null, ?callable $finalize = nul $response = $this->invoke($call->service, $call->method, $context, $request->body); + $headers = array_filter([ + 'headers' => $responseHeaders->count() ? $responseHeaders->packHeaders() : null, + 'trailers' => $responseTrailers->count() ? $responseTrailers->packTrailers() : null, + ]); + $this->workerSend( worker: $worker, body: $response, - headers: Json::encode([ - 'headers' => $responseHeaders->packHeaders(), - 'trailers' => $responseTrailers->packTrailers(), - ]), + headers: !empty($headers) ? Json::encode($headers) : '{}', ); } catch (GRPCExceptionInterface $e) { + $headers = array_filter([ + 'error' => $this->createGrpcError($e), + 'headers' => $responseHeaders->count() ? $responseHeaders->packHeaders() : null, + 'trailers' => $responseTrailers->count() ? $responseTrailers->packTrailers() : null, + ]); + $this->workerSend( worker: $worker, body: '', - headers: Json::encode([ - 'error' => $this->createGrpcError($e), - 'headers' => $responseHeaders->packHeaders(), - 'trailers' => $responseTrailers->packTrailers(), - ]), + headers: Json::encode($headers), ); } catch (\Throwable $e) { $this->workerError($worker, $this->isDebugMode() ? (string) $e : $e->getMessage()); From a7e813f39d7432c5b41019867a9fc71f3b31da4c Mon Sep 17 00:00:00 2001 From: Rauan Mayemir Date: Tue, 10 Sep 2024 15:48:44 +0500 Subject: [PATCH 4/7] Add type-hint for grpc payload header --- src/Internal/CallContext.php | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/src/Internal/CallContext.php b/src/Internal/CallContext.php index 0db223b..7285ef1 100644 --- a/src/Internal/CallContext.php +++ b/src/Internal/CallContext.php @@ -29,6 +29,13 @@ public function __construct( */ public static function decode(string $payload): self { + /** + * @psalm-var array{ + * service: class-string, + * method: non-empty-string, + * context: array> + * } $data + */ $data = Json::decode($payload); return new self( From 1276246877208a2408127d916fb3f2cd2dd2ea9b Mon Sep 17 00:00:00 2001 From: Rauan Mayemir Date: Mon, 13 Jan 2025 14:38:47 +0500 Subject: [PATCH 5/7] Address code review suggestions --- composer.json | 5 ----- src/Internal/CallContext.php | 6 +++--- src/Server.php | 17 ++++++++--------- 3 files changed, 11 insertions(+), 17 deletions(-) diff --git a/composer.json b/composer.json index 41325d1..49f9ef9 100644 --- a/composer.json +++ b/composer.json @@ -77,11 +77,6 @@ "config": { "sort-packages": true }, - "extra": { - "branch-alias": { - "3.x": "3.4.x-dev" - } - }, "minimum-stability": "dev", "prefer-stable": true } diff --git a/src/Internal/CallContext.php b/src/Internal/CallContext.php index 7285ef1..8f4836b 100644 --- a/src/Internal/CallContext.php +++ b/src/Internal/CallContext.php @@ -18,9 +18,9 @@ final class CallContext * @param array> $context */ public function __construct( - public string $service, - public string $method, - public array $context, + public readonly string $service, + public readonly string $method, + public readonly array $context, ) { } diff --git a/src/Server.php b/src/Server.php index 0d883be..ba035eb 100644 --- a/src/Server.php +++ b/src/Server.php @@ -88,22 +88,21 @@ public function serve(?WorkerInterface $worker = null, ?callable $finalize = nul $response = $this->invoke($call->service, $call->method, $context, $request->body); - $headers = array_filter([ - 'headers' => $responseHeaders->count() ? $responseHeaders->packHeaders() : null, - 'trailers' => $responseTrailers->count() ? $responseTrailers->packTrailers() : null, - ]); + $headers = []; + $responseHeaders->count() === 0 or $headers['headers'] = $responseHeaders->packHeaders(); + $responseTrailers->count() === 0 or $headers['trailers'] = $responseTrailers->packTrailers(); $this->workerSend( worker: $worker, body: $response, - headers: !empty($headers) ? Json::encode($headers) : '{}', + headers: $headers === [] ? '{}' : Json::encode($headers), ); } catch (GRPCExceptionInterface $e) { - $headers = array_filter([ + $headers = [ 'error' => $this->createGrpcError($e), - 'headers' => $responseHeaders->count() ? $responseHeaders->packHeaders() : null, - 'trailers' => $responseTrailers->count() ? $responseTrailers->packTrailers() : null, - ]); + ]; + $responseHeaders->count() === 0 or $headers['headers'] = $responseHeaders->packHeaders(); + $responseTrailers->count() === 0 or $headers['trailers'] = $responseTrailers->packTrailers(); $this->workerSend( worker: $worker, From 4c1bcab5bafa5bf508e695a998e44bfa77c70fee Mon Sep 17 00:00:00 2001 From: Rauan Mayemir Date: Mon, 13 Jan 2025 16:27:24 +0500 Subject: [PATCH 6/7] Set RR constraint to ^2024.3 --- composer.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/composer.json b/composer.json index 49f9ef9..7956c91 100644 --- a/composer.json +++ b/composer.json @@ -45,7 +45,7 @@ "google/protobuf": "^3.7 || ^4.0", "spiral/roadrunner-worker": "^3.0", "spiral/goridge": "^4.0", - "spiral/roadrunner": "^2023.1 || ^2024.1" + "spiral/roadrunner": "^2024.3" }, "require-dev": { "jetbrains/phpstorm-attributes": "^1.0", From 5a6ea307b6ff34d297a88bbdf9fac62898cfc5ee Mon Sep 17 00:00:00 2001 From: Rauan Mayemir Date: Mon, 13 Jan 2025 18:03:38 +0500 Subject: [PATCH 7/7] Add test coverage for invoke exception with outgoing headers and trailers --- psalm-baseline.xml | 2 +- src/Internal/CallContext.php | 3 +-- src/ResponseTrailers.php | 3 +-- src/Server.php | 6 +++--- tests/ContextTest.php | 21 ++++++++++++++++++++ tests/ServerTest.php | 37 +++++++++++++++++++++++++++++++++++- tests/Stub/TestService.php | 12 ++++++++++++ 7 files changed, 75 insertions(+), 9 deletions(-) diff --git a/psalm-baseline.xml b/psalm-baseline.xml index 119d221..5c064bb 100644 --- a/psalm-baseline.xml +++ b/psalm-baseline.xml @@ -1,2 +1,2 @@ - + diff --git a/src/Internal/CallContext.php b/src/Internal/CallContext.php index 8f4836b..5f6ce29 100644 --- a/src/Internal/CallContext.php +++ b/src/Internal/CallContext.php @@ -21,8 +21,7 @@ public function __construct( public readonly string $service, public readonly string $method, public readonly array $context, - ) { - } + ) {} /** * @throws \JsonException diff --git a/src/ResponseTrailers.php b/src/ResponseTrailers.php index 3231013..d0e4fca 100644 --- a/src/ResponseTrailers.php +++ b/src/ResponseTrailers.php @@ -39,10 +39,9 @@ public function set(string $key, string $value): void /** * @param THeaderKey $key - * @param string|null $default * @return THeaderValue|null */ - public function get(string $key, string $default = null): ?string + public function get(string $key, ?string $default = null): ?string { return $this->trailers[$key] ?? $default; } diff --git a/src/Server.php b/src/Server.php index ba035eb..78607fc 100644 --- a/src/Server.php +++ b/src/Server.php @@ -78,12 +78,12 @@ public function serve(?WorkerInterface $worker = null, ?callable $finalize = nul try { $call = CallContext::decode($request->header); - $context = new Context(array_merge( + $context = new Context(\array_merge( $call->context, [ ResponseHeaders::class => $responseHeaders, - ResponseTrailers::class => $responseTrailers - ] + ResponseTrailers::class => $responseTrailers, + ], )); $response = $this->invoke($call->service, $call->method, $context, $request->body); diff --git a/tests/ContextTest.php b/tests/ContextTest.php index 88b360e..ee79f95 100644 --- a/tests/ContextTest.php +++ b/tests/ContextTest.php @@ -7,6 +7,7 @@ use PHPUnit\Framework\TestCase; use Spiral\RoadRunner\GRPC\Context; use Spiral\RoadRunner\GRPC\ResponseHeaders; +use Spiral\RoadRunner\GRPC\ResponseTrailers; class ContextTest extends TestCase { @@ -75,4 +76,24 @@ public function testGetOutgoingHeaders(): void $ctx = new Context([ResponseHeaders::class => $outgoingHeaders]); $this->assertSame($outgoingHeaders, $ctx->getValue(ResponseHeaders::class)); } + + public function testGetOutgoingTrailer(): void + { + $outgoingTrailers = [ + 'X-Some-Trailer' => 'foobar', + ]; + $ctx = new Context([ResponseTrailers::class => new ResponseTrailers($outgoingTrailers)]); + + $this->assertSame($outgoingTrailers['X-Some-Trailer'], $ctx->getValue(ResponseTrailers::class)->get('X-Some-Trailer')); + $this->assertNull($ctx->getValue(ResponseTrailers::class)->get('not-existing')); + } + + public function testGetOutgoingTrailers(): void + { + $outgoingTrailers = new ResponseTrailers([ + 'X-Some-Trailer' => 'foobar', + ]); + $ctx = new Context([ResponseTrailers::class => $outgoingTrailers]); + $this->assertSame($outgoingTrailers, $ctx->getValue(ResponseTrailers::class)); + } } diff --git a/tests/ServerTest.php b/tests/ServerTest.php index 7d93822..b375df3 100644 --- a/tests/ServerTest.php +++ b/tests/ServerTest.php @@ -7,17 +7,18 @@ use Google\Rpc\Status; use Mockery as m; use PHPUnit\Framework\TestCase; +use Service\DetailsMessageForException; use Service\Message; use Service\TestInterface; use Spiral\Goridge\Frame; use Spiral\Goridge\RelayInterface; use Spiral\RoadRunner\GRPC\Exception\ServiceException; +use Spiral\RoadRunner\GRPC\InvokerInterface; use Spiral\RoadRunner\GRPC\Server; use Spiral\RoadRunner\GRPC\Tests\Stub\TestService; use Spiral\RoadRunner\Payload; use Spiral\RoadRunner\Worker; use Spiral\RoadRunner\WorkerInterface; -use Spiral\RoadRunner\GRPC\InvokerInterface; class ServerTest extends TestCase { @@ -143,6 +144,40 @@ public function testExceptionDetails(): void $server->serve($worker); } + public function testInvokeGrpcException(): void + { + $worker = m::mock(WorkerInterface::class); + $worker->shouldReceive('waitPayload') + ->times(2) + ->andReturn( + new Payload( + body: $this->packMessage('withDetailsAndHeaders'), + header: '{"context": {}, "service": "service.Test", "method": "Throw"}', + ), + null, + ); + + $worker->shouldReceive('respond')->once() + ->withArgs(static function (Payload $payload) { + $header = \json_decode($payload->header, true); + + $status = new Status(); + $status->mergeFromString(\base64_decode($header['error'])); + /** @var DetailsMessageForException $message */ + $message = $status->getDetails()->offsetGet(0)->unpack(); + + $outgoingHeaders = \json_decode($header['headers'], true); + $outgoingTrailers = \json_decode($header['trailers'], true); + + return $message instanceof DetailsMessageForException + && $message->getMessage() === 'details message' + && $outgoingHeaders === ['foo' => 'bar'] + && $outgoingTrailers === ['baz' => 'bar']; + }); + + $this->server->serve($worker); + } + protected function setUp(): void { parent::setUp(); diff --git a/tests/Stub/TestService.php b/tests/Stub/TestService.php index 3c10674..b98d914 100644 --- a/tests/Stub/TestService.php +++ b/tests/Stub/TestService.php @@ -34,6 +34,17 @@ public function Throw(ContextInterface $ctx, Message $in): Message $grpcException = new GRPCException("main exception message", 3, [$detailsMessage]); + throw $grpcException; + case "withDetailsAndHeaders": + $ctx->getValue(GRPC\ResponseHeaders::class)->set('foo', 'bar'); + $ctx->getValue(GRPC\ResponseTrailers::class)->set('baz', 'bar'); + + $detailsMessage = new DetailsMessageForException(); + $detailsMessage->setCode(1); + $detailsMessage->setMessage("details message"); + + $grpcException = new GRPCException("main exception message", 3, [$detailsMessage]); + throw $grpcException; case "regularException": { @@ -70,6 +81,7 @@ public function Info(ContextInterface $ctx, Message $in): Message } $ctx->getValue(GRPC\ResponseHeaders::class)->set('foo', 'bar'); + $ctx->getValue(GRPC\ResponseTrailers::class)->set('baz', 'bar'); return $out; }