From f67f1e6d7e938304e0353a54b93ba129e15cf044 Mon Sep 17 00:00:00 2001 From: kafkiansky Date: Thu, 15 Oct 2026 20:17:32 +0300 Subject: [PATCH] Add micro service errors, handlers and drain --- README.md | 143 ++++++ src/Exception/ServiceAlreadyStopped.php | 12 + src/Micro/EndpointStats.php | 5 +- src/Micro/Group.php | 5 + src/Micro/Internal/EndpointHandler.php | 57 ++- src/Micro/Internal/RequestOutcome.php | 25 ++ src/Micro/Internal/StatsResponse.php | 15 +- src/Micro/Request.php | 32 +- src/Micro/Response.php | 8 +- src/Micro/Service.php | 155 ++++++- src/Micro/Service/State.php | 22 + src/Micro/ServiceConfig.php | 6 + src/Micro/ServiceError.php | 26 ++ tests/Micro/ServiceTest.php | 566 +++++++++++++++++++++++- 14 files changed, 1038 insertions(+), 39 deletions(-) create mode 100644 src/Exception/ServiceAlreadyStopped.php create mode 100644 src/Micro/Internal/RequestOutcome.php create mode 100644 src/Micro/Service/State.php create mode 100644 src/Micro/ServiceError.php diff --git a/README.md b/README.md index 34a261c..c9ab1d0 100644 --- a/README.md +++ b/README.md @@ -44,6 +44,9 @@ Pure non-blocking (fiber based) strictly typed full-featured PHP driver for NATS - [Micro Service](#micro-service) - [Endpoints](#service-endpoints) - [Groups](#groups) + - [Errors](#service-errors) + - [Custom stats](#custom-stats) + - [Stopping a service](#stopping-a-service) ## Installation @@ -1528,6 +1531,146 @@ $nc dump($nc->request('srv.api.v1.echo', new Nats\Message('ping'))->message->payload); ``` +#### Service errors + +`Request::error()` replies with the `Nats-Service-Error-Code` and `Nats-Service-Error` headers and an optional payload, and counts the request as failed in the endpoint stats (`num_errors`, `last_error`). + +```php +createService(new Micro\ServiceConfig('UserService', '1.0.0')); + +$srv + ->addEndpoint(new Micro\EndpointConfig('find', subject: 'users.find'), static function (Micro\Request $request): void { + $request->error(404, 'User not found', '{"id": 42}'); + }); + +$response = $nc->request('users.find', new Nats\Message('42')); + +dump( + $response->message->headers?->get(Nats\Header\ServiceErrorCode::header()), // 404 + $response->message->headers?->get(Nats\Header\ServiceError::header()), // User not found + $response->message->payload, // {"id": 42} +); +``` + +A handler can also throw a `Micro\ServiceError` to reply with its code, description and payload. +Any other exception thrown before the handler has replied is answered with code `500` and the exception message as the description, so the requester does not wait for its timeout. +Either way the error the requester got is counted in the endpoint stats: `lastError` is that `Micro\ServiceError`, with the original exception as the previous one, and `last_error` in `$SRV.STATS` reads e.g. `500:`. + +```php +createService(new Micro\ServiceConfig('UserService', '1.0.0')); + +$srv + ->addEndpoint(new Micro\EndpointConfig('create', subject: 'users.create'), static function (Micro\Request $request): void { + throw new Micro\ServiceError(409, 'User already exists'); + }); + +$response = $nc->request('users.create', new Nats\Message('{"name": "John"}')); + +dump($response->message->headers?->get(Nats\Header\ServiceErrorCode::header())); // 409 +``` + +#### Custom stats + +The `statsHandler` returns the custom `data` of every endpoint's stats. The value is encoded as JSON in the `$SRV.STATS` responses and is available in `Service::stats()` as is. + +```php +createService(new Micro\ServiceConfig( + name: 'EchoService', + version: '1.0.0', + statsHandler: static fn(Micro\EndpointInfo $endpoint): array => ['endpoint' => $endpoint->name, 'memory' => memory_get_usage()], +)); + +$srv->addEndpoint(new Micro\EndpointConfig('echo'), static function (Micro\Request $request): void { + $request->respond(new Micro\Response($request->data)); +}); + +dump($srv->stats()->endpoints[0]->data); +``` + +#### Stopping a service + +`Service::stop()` unsubscribes the endpoints at once and drops the requests they have not handled yet, `Service::drain()` unsubscribes them and waits until the requests already received are handled. +The service is also stopped when its subscriptions are completed from the outside: when the client is stopped, or when the server rejects a subscription, e.g. on a permissions violation. +Once the service has started stopping, `addEndpoint()` (on the service or a group) throws `Nats\Exception\ServiceAlreadyStopped`. +`Service::$state` transitions from `Micro\Service\State::Alive` to `Stopping` as soon as shutdown starts, and to `Stopped` once shutdown completes. +New endpoints and groups cannot be added while the service is stopping or after it has stopped: these calls throw `ServiceAlreadyStopped`. + +The `doneHandler` is called asynchronously once the service has stopped, for any of these reasons. The `errorHandler` is called asynchronously with the error of a failed service subscription, after which the service is stopped. +Subscription failures are reported once, for the first failed subscription only: e.g. a permissions violation fails all the `$SRV` subscriptions at once, but the service is already stopping after the first one (nats.go calls its handler for every failed subscription). +Exceptions from `doneHandler` are also passed to `errorHandler`. If no `errorHandler` is configured, or it throws an exception itself, the exception goes to the Revolt event loop error handler (`EventLoop::setErrorHandler()`). +Callback failures do not change the service's `Stopped` state. Callbacks may call `stop()` or `drain()` again. + +```php +createService(new Micro\ServiceConfig( + name: 'EchoService', + version: '1.0.0', + doneHandler: static function (Micro\Service $srv): void { + dump("Service {$srv->info()->identity->id} has stopped."); + }, + errorHandler: static function (Micro\Service $srv, \Throwable $e): void { + dump("Service {$srv->info()->identity->id} has failed: {$e->getMessage()}"); + }, +)); + +$srv->addEndpoint(new Micro\EndpointConfig('echo'), static function (Micro\Request $request): void { + $request->respond(new Micro\Response($request->data)); +}); + +$srv->drain(); +``` + ## License The MIT License (MIT). Please see [License File](LICENSE) for more information. diff --git a/src/Exception/ServiceAlreadyStopped.php b/src/Exception/ServiceAlreadyStopped.php new file mode 100644 index 0000000..a9f99ff --- /dev/null +++ b/src/Exception/ServiceAlreadyStopped.php @@ -0,0 +1,12 @@ +svc->addGroup( diff --git a/src/Micro/Internal/EndpointHandler.php b/src/Micro/Internal/EndpointHandler.php index 5a1737d..0883af6 100644 --- a/src/Micro/Internal/EndpointHandler.php +++ b/src/Micro/Internal/EndpointHandler.php @@ -9,12 +9,15 @@ use Thesis\Nats\Micro\EndpointInfo; use Thesis\Nats\Micro\EndpointStats; use Thesis\Nats\Micro\Request; +use Thesis\Nats\Micro\ServiceError; /** * @internal */ final class EndpointHandler { + private const int INTERNAL_ERROR_CODE = 500; + /** @var non-negative-int */ private int $requests = 0; @@ -28,26 +31,43 @@ final class EndpointHandler /** * @param callable(Request): void $handler + * @param ?\Closure(EndpointInfo): mixed $statsHandler */ public function __construct( public readonly EndpointInfo $info, private readonly mixed $handler, private readonly Encoder $encoder, + private readonly ?\Closure $statsHandler = null, ) {} public function handle(Delivery $delivery): void { $start = (int) hrtime(true); + $outcome = new RequestOutcome(); + $request = new Request($delivery, $this->encoder, $outcome); try { - ($this->handler)(new Request($delivery, $this->encoder)); + ($this->handler)($request); } catch (\Throwable $e) { - ++$this->errors; - $this->lastError = $e; - } finally { - ++$this->requests; - $this->processingTime += max(0, (int) hrtime(true) - $start); + // A failure after the handler has replied is counted as is: the requester has got its response. + $outcome->fail(!$outcome->replied && $request->reply !== null ? self::replyWithError($request, $e) : $e); } + + ++$this->requests; + $this->processingTime += max(0, (int) hrtime(true) - $start); + + if ($outcome->error !== null) { + $this->fail($outcome->error); + } + } + + /** + * Counts an error that is not tied to a request, e.g. the endpoint subscription failure. + */ + public function fail(\Throwable $error): void + { + ++$this->errors; + $this->lastError = $error; } public function reset(): void @@ -69,6 +89,31 @@ public function stats(): EndpointStats lastError: $this->lastError, processingTime: $this->processingTime, averageProcessingTime: $this->requests > 0 ? max(0, intdiv($this->processingTime, $this->requests)) : 0, + data: $this->statsHandler !== null ? ($this->statsHandler)($this->info) : null, ); } + + /** + * A handler that throws still answers the requester instead of leaving it waiting for the timeout: + * a {@see ServiceError} with its own code, anything else with 500 and the exception message. + * + * @return ServiceError the error replied with, counted in the stats so that they match the reply + */ + private static function replyWithError(Request $request, \Throwable $e): ServiceError + { + if ($e instanceof ServiceError) { + $error = $e; + } else { + $message = $e->getMessage(); + $error = new ServiceError(self::INTERNAL_ERROR_CODE, $message !== '' ? $message : $e::class, previous: $e); + } + + try { + $request->error($error->getCode(), $error->description, $error->data); + } catch (\Throwable) { + // The reply cannot be sent, e.g. the connection is closed: the handler's error is the one to report. + } + + return $error; + } } diff --git a/src/Micro/Internal/RequestOutcome.php b/src/Micro/Internal/RequestOutcome.php new file mode 100644 index 0000000..a6aeb1d --- /dev/null +++ b/src/Micro/Internal/RequestOutcome.php @@ -0,0 +1,25 @@ +replied = true; + } + + public function fail(\Throwable $error): void + { + $this->error = $error; + } +} diff --git a/src/Micro/Internal/StatsResponse.php b/src/Micro/Internal/StatsResponse.php index d202dae..7112152 100644 --- a/src/Micro/Internal/StatsResponse.php +++ b/src/Micro/Internal/StatsResponse.php @@ -5,6 +5,7 @@ namespace Thesis\Nats\Micro\Internal; use Thesis\Nats\Micro\EndpointStats; +use Thesis\Nats\Micro\ServiceError; use Thesis\Nats\Micro\Stats; /** @@ -35,7 +36,7 @@ public function jsonSerialize(): array 'queue_group' => $stats->queueGroup, 'num_requests' => $stats->numRequests, 'num_errors' => $stats->numErrors, - 'last_error' => $stats->lastError?->getMessage(), + 'last_error' => self::formatError($stats->lastError), 'processing_time' => $stats->processingTime, 'average_processing_time' => $stats->averageProcessingTime, 'data' => $stats->data, @@ -44,4 +45,16 @@ public function jsonSerialize(): array ), ]; } + + /** + * Formats a service error as "code:description", the same way nats.go does. + */ + private static function formatError(?\Throwable $error): ?string + { + if ($error instanceof ServiceError) { + return "{$error->getCode()}:{$error->description}"; + } + + return $error?->getMessage(); + } } diff --git a/src/Micro/Request.php b/src/Micro/Request.php index c5c9bbc..c1b8f90 100644 --- a/src/Micro/Request.php +++ b/src/Micro/Request.php @@ -25,9 +25,13 @@ /** @var ?non-empty-string */ public ?string $reply; + /** + * @internal + */ public function __construct( private Delivery $delivery, private Encoder $encoder, + private Internal\RequestOutcome $outcome = new Internal\RequestOutcome(), ) { $this->data = $this->delivery->message->payload; $this->headers = $this->delivery->message->headers ?? new Headers(); @@ -37,14 +41,38 @@ public function __construct( public function respond(Response $response, ?Cancellation $cancellation = null): void { + $this->outcome->markReplied(); $this->delivery->reply(new Message($response->data, $response->headers), $cancellation); } public function respondJson(mixed $response, ?Cancellation $cancellation = null): void { - $this->delivery->reply( - new Message($this->encoder->encode($response)), + // Encoded first: a response that cannot be encoded is not sent, so the requester still gets the error reply. + $message = new Message($this->encoder->encode($response)); + + $this->outcome->markReplied(); + $this->delivery->reply($message, $cancellation); + } + + /** + * Replies with the Nats-Service-Error and Nats-Service-Error-Code headers and counts the request as failed in the endpoint stats. + * + * @param non-empty-string $description + */ + public function error( + int $code, + string $description, + ?string $data = null, + Headers $headers = new Headers(), + ?Cancellation $cancellation = null, + ): void { + $this->respond( + new Response($data, $headers) + ->withErrorCode($code) + ->withErrorDescription($description), $cancellation, ); + + $this->outcome->fail(new ServiceError($code, $description, $data)); } } diff --git a/src/Micro/Response.php b/src/Micro/Response.php index 63dd699..7fde2ec 100644 --- a/src/Micro/Response.php +++ b/src/Micro/Response.php @@ -21,7 +21,9 @@ public function withErrorCode(int $code): self { return new self( data: $this->data, - headers: $this->headers->with(Header\ServiceErrorCode::header(), $code), + headers: $this->headers + ->without(Header\ServiceErrorCode::header()) + ->with(Header\ServiceErrorCode::header(), $code), ); } @@ -32,7 +34,9 @@ public function withErrorDescription(string $description): self { return new self( data: $this->data, - headers: $this->headers->with(Header\ServiceError::header(), $description), + headers: $this->headers + ->without(Header\ServiceError::header()) + ->with(Header\ServiceError::header(), $description), ); } } diff --git a/src/Micro/Service.php b/src/Micro/Service.php index 21e8365..e8526cd 100644 --- a/src/Micro/Service.php +++ b/src/Micro/Service.php @@ -5,11 +5,15 @@ namespace Thesis\Nats\Micro; use Amp\Cancellation; +use Amp\Future; +use Revolt\EventLoop; use Thesis\Nats\Client; use Thesis\Nats\Delivery; +use Thesis\Nats\Exception\ServiceAlreadyStopped; use Thesis\Nats\Json\Encoder; use Thesis\Nats\NatsException; use Thesis\Nats\Subscription; +use function Amp\async; /** * @api @@ -32,6 +36,11 @@ final class Service /** @var non-empty-string */ private readonly string $queue; + /** @var ?Future */ + private ?Future $shutdown = null; + + public private(set) Service\State $state = Service\State::Alive; + public function __construct( private readonly Client $nc, private readonly ServiceIdentity $identity, @@ -48,8 +57,15 @@ public function __construct( $this->registerVerbHandler(Internal\Verb::Stats, $this->handleStats(...)); } + /** + * @throws ServiceAlreadyStopped once the service has started stopping + */ public function addGroup(GroupConfig $config): Group { + if ($this->stopping()) { + throw new ServiceAlreadyStopped(); + } + return $this->groups[$config->name] ??= new Group( svc: $this, name: $config->name, @@ -59,6 +75,7 @@ public function addGroup(GroupConfig $config): Group /** * @param callable(Request): void $handler + * @throws ServiceAlreadyStopped once the service has started stopping * @throws NatsException */ public function addEndpoint( @@ -66,6 +83,10 @@ public function addEndpoint( callable $handler, ?Cancellation $cancellation = null, ): self { + if ($this->stopping()) { + throw new ServiceAlreadyStopped(); + } + $endpointHandler = new Internal\EndpointHandler( info: new EndpointInfo( name: $config->name, @@ -75,6 +96,7 @@ public function addEndpoint( ), handler: $handler, encoder: $this->encoder, + statsHandler: $this->config->statsHandler, ); $subscription = $this->nc->subscribe( @@ -84,7 +106,15 @@ public function addEndpoint( cancellation: $cancellation, ); + // The service has started stopping while subscribing: the stop no longer sees the endpoint, so it must not outlive the service. + if ($this->stopping()) { + $subscription->stop(); + + throw new ServiceAlreadyStopped(); + } + $this->endpoints[] = new Internal\SubscribedEndpoint($endpointHandler, $subscription); + $this->supervise($subscription, $endpointHandler); return $this; } @@ -113,11 +143,33 @@ public function stats(): Stats ); } + /** + * Unsubscribes the endpoints at once, dropping the requests they have not handled yet. + * The cancellation stops waiting only: the service is stopped to the end anyway. + */ public function stop(?Cancellation $cancellation = null): void { - $this->stopEndpoints($cancellation); - $this->stopInternalEndpoints($cancellation); - $this->stopGroups(); + $this->shutdown(static fn(Subscription $subscription) => $subscription->stop(), $cancellation); + } + + /** + * Unsubscribes the endpoints and waits for them to handle the requests already received. + * The cancellation stops waiting only: the service keeps draining and is stopped once the handlers are done. + */ + public function drain(?Cancellation $cancellation = null): void + { + $this->shutdown( + static function (Subscription $subscription): void { + $subscription->drain(); + + try { + $subscription->awaitCompletion(); + } catch (\Throwable) { + // The subscription has failed while draining, e.g. the connection is lost: the service is going down anyway. + } + }, + $cancellation, + ); } public function reset(): void @@ -147,7 +199,7 @@ private function registerInternalHandler( callable $handler, ): void { $encoder = $this->encoder; - $this->verbs[$name] = $this->nc->subscribe( + $this->verbs[$name] = $subscription = $this->nc->subscribe( $subject->value, static function (Delivery $delivery) use ( $handler, @@ -156,6 +208,8 @@ static function (Delivery $delivery) use ( $handler(new Request($delivery, $encoder)); }, ); + + $this->supervise($subscription); } private function handlePing(Request $request): void @@ -173,28 +227,97 @@ private function handleStats(Request $request): void $request->respondJson(new Internal\StatsResponse($this->stats())); } - private function stopEndpoints(?Cancellation $cancellation = null): void + /** + * Stops the service once a subscription of it completes from the outside: the client is stopped, + * or the server rejected the subscription, which is reported to the error handler. + */ + private function supervise(Subscription $subscription, ?Internal\EndpointHandler $endpoint = null): void { - $endpoints = $this->endpoints; - $this->endpoints = []; + $future = async(function () use ( + $subscription, + $endpoint, + ): void { + try { + $subscription->awaitCompletion(); + } catch (\Throwable $e) { + if (!$this->stopping()) { + $endpoint?->fail($e); + $this->reportError($e); + } + } + + if (!$this->stopping()) { + $this->stop(); + } + }); + $future->ignore(); + } - foreach ($endpoints as $it) { - $it->subscription->stop($cancellation); + private function stopping(): bool + { + return $this->state !== Service\State::Alive; + } + + /** + * @param \Closure(Subscription): void $complete + */ + private function shutdown(\Closure $complete, ?Cancellation $cancellation = null): void + { + if ($this->shutdown === null) { + $this->state = Service\State::Stopping; + $this->shutdown = async($this->complete(...), $complete); } + + $this->shutdown->await($cancellation); } - private function stopInternalEndpoints(?Cancellation $cancellation = null): void + /** + * @param \Closure(Subscription): void $complete + */ + private function complete(\Closure $complete): void + { + try { + // The endpoints go first, so that the service still answers PING, INFO and STATS while it drains. + Future\await(array_map( + static fn(Internal\SubscribedEndpoint $it): Future => async($complete, $it->subscription), + $this->endpoints, + )); + + Future\await(array_map( + static fn(Subscription $it): Future => async($complete, $it), + $this->verbs, + )); + } finally { + $this->endpoints = []; + $this->verbs = []; + $this->groups = []; + $this->state = Service\State::Stopped; + + if ($this->config->doneHandler !== null) { + EventLoop::queue($this->callDoneHandler(...)); + } + } + } + + private function callDoneHandler(): void { - $verbs = $this->verbs; - $this->verbs = []; + try { + if ($this->config->doneHandler !== null) { + ($this->config->doneHandler)($this); + } + } catch (\Throwable $e) { + if ($this->config->errorHandler === null) { + throw $e; + } - foreach ($verbs as $sub) { - $sub->stop($cancellation); + $this->reportError($e); } } - private function stopGroups(): void + private function reportError(\Throwable $e): void { - $this->groups = []; + if ($this->config->errorHandler !== null) { + EventLoop::queue($this->config->errorHandler, $this, $e); + } } } diff --git a/src/Micro/Service/State.php b/src/Micro/Service/State.php new file mode 100644 index 0000000..3fe5852 --- /dev/null +++ b/src/Micro/Service/State.php @@ -0,0 +1,22 @@ + $metadata * @param ?non-empty-string $queueGroup + * @param ?\Closure(EndpointInfo): mixed $statsHandler returns the custom `data` of the endpoint stats, encoded as JSON + * @param ?\Closure(Service): void $doneHandler called asynchronously once the service has stopped; exceptions go to $errorHandler, or to the Revolt error handler if none is configured + * @param ?\Closure(Service, \Throwable): void $errorHandler called asynchronously for the first failed subscription and for exceptions from $doneHandler; exceptions from this callback go to the Revolt error handler */ public function __construct( public string $name, @@ -26,6 +29,9 @@ public function __construct( public string $description = '', public array $metadata = [], public ?string $queueGroup = null, + public ?\Closure $statsHandler = null, + public ?\Closure $doneHandler = null, + public ?\Closure $errorHandler = null, ) { if (preg_match(self::NAME_REGEX, $this->name) !== 1) { throw new \InvalidArgumentException('Service name is not a valid string (only "A-Z, a-z, 0-9, _, -" are allowed).'); diff --git a/src/Micro/ServiceError.php b/src/Micro/ServiceError.php new file mode 100644 index 0000000..705c613 --- /dev/null +++ b/src/Micro/ServiceError.php @@ -0,0 +1,26 @@ +request('echo', new Message('Ping')); self::assertSame('Ping', $response->message->payload); - $deferred = new DeferredFuture(); - - try { - $nc->request('error', new Message('Ping'), new TimeoutCancellation(0.5)); - } catch (CancelledException) { - $deferred->complete(); - } - - $deferred->getFuture()->await(); + $response = $nc->request('error', new Message('Ping'), new TimeoutCancellation(5)); + self::assertSame(500, $response->message->headers?->get(Header\ServiceErrorCode::header())); + self::assertSame('Invalid request', $response->message->headers->get(Header\ServiceError::header())); $stats = $srv->stats(); self::assertCount(2, $stats->endpoints); @@ -110,7 +116,14 @@ public function testServiceStats(): void self::assertGreaterThan(1_000, $stats->endpoints[1]->averageProcessingTime); self::assertSame(1, $stats->endpoints[1]->numRequests); self::assertSame(1, $stats->endpoints[1]->numErrors); - self::assertSame('Invalid request', $stats->endpoints[1]->lastError?->getMessage()); + self::assertInstanceOf(ServiceError::class, $stats->endpoints[1]->lastError); + self::assertSame(500, $stats->endpoints[1]->lastError->getCode()); + self::assertSame('Invalid request', $stats->endpoints[1]->lastError->description); + self::assertInstanceOf(\LogicException::class, $stats->endpoints[1]->lastError->getPrevious()); + + /** @var array{endpoints: list} $raw */ + $raw = json_decode((string) $nc->request("\$SRV.STATS.EchoService.{$srv->info()->identity->id}")->message->payload, true, flags: JSON_THROW_ON_ERROR); + self::assertSame('500:Invalid request', $raw['endpoints'][1]['last_error'] ?? null); $srv->stop(); } @@ -192,6 +205,539 @@ public function testMetadataIsEncodedAsJsonObject(): void $srv->stop(); } + public function testRequestError(): void + { + $nc = $this->client(); + $id = Id\generateUniqueId(10); + + $srv = $nc->createService(new ServiceConfig( + name: 'ErrorService', + version: '1.0.0', + )); + + $srv->addEndpoint(new EndpointConfig('find', subject: "errors.{$id}"), static function (Request $request): void { + $request->error(404, 'Not Found', '{"id":42}', new Headers()->with(Header\ServiceErrorCode::header(), 400)); + }); + + $response = $nc->request("errors.{$id}", new Message('Ping'), new TimeoutCancellation(5)); + + self::assertSame('{"id":42}', $response->message->payload); + self::assertSame([404], $response->message->headers?->values(Header\ServiceErrorCode::header())); + self::assertSame('Not Found', $response->message->headers->get(Header\ServiceError::header())); + + $stats = $srv->stats()->endpoints; + self::assertCount(1, $stats); + $stats = $stats[0]; + self::assertSame(1, $stats->numRequests); + self::assertSame(1, $stats->numErrors); + self::assertInstanceOf(ServiceError::class, $stats->lastError); + self::assertSame(404, $stats->lastError->getCode()); + self::assertSame('Not Found', $stats->lastError->description); + + /** @var array{endpoints: list} $raw */ + $raw = json_decode((string) $nc->request("\$SRV.STATS.ErrorService.{$srv->info()->identity->id}")->message->payload, true, flags: JSON_THROW_ON_ERROR); + self::assertSame('404:Not Found', $raw['endpoints'][0]['last_error'] ?? null); + + $srv->stop(); + } + + public function testThrownServiceErrorIsReplied(): void + { + $nc = $this->client(); + $id = Id\generateUniqueId(10); + + $srv = $nc->createService(new ServiceConfig( + name: 'ErrorService', + version: '1.0.0', + )); + + $srv->addEndpoint(new EndpointConfig('find', subject: "errors.{$id}"), static function (Request $_): void { + throw new ServiceError(409, 'Conflict', 'details'); + }); + + $response = $nc->request("errors.{$id}", new Message('Ping'), new TimeoutCancellation(5)); + + self::assertSame('details', $response->message->payload); + self::assertSame(409, $response->message->headers?->get(Header\ServiceErrorCode::header())); + self::assertSame('Conflict', $response->message->headers->get(Header\ServiceError::header())); + $stats = $srv->stats()->endpoints; + self::assertCount(1, $stats); + self::assertSame(1, $stats[0]->numErrors); + + $srv->stop(); + } + + public function testHandlerFailureAfterReplyIsCountedButNotRepliedAgain(): void + { + $nc = $this->client(); + $id = Id\generateUniqueId(10); + + $srv = $nc->createService(new ServiceConfig( + name: 'ErrorService', + version: '1.0.0', + )); + + $srv->addEndpoint(new EndpointConfig('echo', subject: "errors.{$id}"), static function (Request $request): void { + $request->respond(new Response($request->data)); + + throw new \LogicException('Failed after reply'); + }); + + /** @var \ArrayObject $replies */ + $replies = new \ArrayObject(); + $inbox = "_INBOX.{$id}"; + $nc->subscribe($inbox, static function (Delivery $delivery) use ($replies): void { + $replies[] = $delivery->message; + }); + $nc->publish("errors.{$id}", new Message('Ping'), replyTo: $inbox); + + delay(0.3); + + self::assertSame(['Ping'], array_map(static fn(Message $reply): ?string => $reply->payload, [...$replies])); + self::assertFalse($replies[0]?->headers?->exists(Header\ServiceErrorCode::header()) ?? false); + + $stats = $srv->stats()->endpoints; + self::assertCount(1, $stats); + $stats = $stats[0]; + self::assertSame(1, $stats->numRequests); + self::assertSame(1, $stats->numErrors); + // No error reply has been sent, so the failure is counted as is, not as a 500 service error. + self::assertInstanceOf(\LogicException::class, $stats->lastError); + self::assertSame('Failed after reply', $stats->lastError->getMessage()); + + $srv->stop(); + } + + public function testStatsHandler(): void + { + $nc = $this->client(); + $id = Id\generateUniqueId(10); + + $srv = $nc->createService(new ServiceConfig( + name: 'StatsHandlerService', + version: '1.0.0', + statsHandler: static fn(EndpointInfo $info): array => ['endpoint' => $info->name, 'subject' => $info->subject], + )); + + $srv->addEndpoint(new EndpointConfig('echo', subject: "stats.{$id}"), static function (Request $request): void { + $request->respond(new Response($request->data)); + }); + + $stats = $srv->stats()->endpoints; + self::assertCount(1, $stats); + self::assertSame(['endpoint' => 'echo', 'subject' => "stats.{$id}"], $stats[0]->data); + + $payload = (string) $nc->request("\$SRV.STATS.StatsHandlerService.{$srv->info()->identity->id}")->message->payload; + self::assertStringContainsString("\"data\":{\"endpoint\":\"echo\",\"subject\":\"stats.{$id}\"}", $payload); + + $srv->stop(); + } + + public function testStatsWithoutStatsHandlerHaveNoData(): void + { + $nc = $this->client(); + + $srv = $nc->createService(new ServiceConfig( + name: 'NoStatsHandlerService', + version: '1.0.0', + )); + + $srv->addEndpoint(new EndpointConfig('echo', subject: 'no-stats.' . Id\generateUniqueId(10)), static fn() => null); + + $stats = $srv->stats()->endpoints; + self::assertCount(1, $stats); + self::assertNull($stats[0]->data); + + $payload = (string) $nc->request("\$SRV.STATS.NoStatsHandlerService.{$srv->info()->identity->id}")->message->payload; + self::assertStringNotContainsString('"data"', $payload); + + $srv->stop(); + } + + public function testDrainHandlesReceivedRequests(): void + { + $nc = $this->client(); + $id = Id\generateUniqueId(10); + + /** @var DeferredFuture $started */ + $started = new DeferredFuture(); + + /** @var DeferredFuture $release */ + $release = new DeferredFuture(); + + /** @var \ArrayObject $done */ + $done = new \ArrayObject(); + + $srv = $nc->createService(new ServiceConfig( + name: 'DrainService', + version: '1.0.0', + doneHandler: static function (Service $srv) use ($done): void { + $done[] = $srv; + }, + )); + + $srv->addEndpoint(new EndpointConfig('slow', subject: "drain.{$id}"), static function (Request $request) use ( + $started, + $release, + ): void { + if (!$started->isComplete()) { + $started->complete(); + } + + $release->getFuture()->await(); + $request->respond(new Response("done {$request->data}")); + }); + + $replies = []; + for ($i = 0; $i < 3; ++$i) { + $replies[] = async($nc->request(...), "drain.{$id}", new Message((string) $i), cancellation: new TimeoutCancellation(5)); + } + + $started->getFuture()->await(new TimeoutCancellation(5)); + + // Let the other requests reach the endpoint's buffer. + delay(0.2); + + $drain = async($srv->drain(...)); + delay(0.1); + self::assertFalse($drain->isComplete()); + self::assertSame(Service\State::Stopping, $srv->state); + + $release->complete(); + $drain->await(new TimeoutCancellation(5)); + + self::assertSame(Service\State::Stopped, $srv->state); + self::assertSame( + ['done 0', 'done 1', 'done 2'], + array_map(static fn(Delivery $delivery): ?string => $delivery->message->payload, Future\await($replies)), + ); + + delay(0.05); + self::assertCount(1, $done); + self::assertSame($srv, $done[0]); + } + + public function testCancelledDrainStillStopsTheService(): void + { + $nc = $this->client(); + $id = Id\generateUniqueId(10); + + /** @var DeferredFuture $started */ + $started = new DeferredFuture(); + + /** @var DeferredFuture $release */ + $release = new DeferredFuture(); + + /** @var DeferredFuture $done */ + $done = new DeferredFuture(); + + $srv = $nc->createService(new ServiceConfig( + name: 'CancelledDrainService', + version: '1.0.0', + doneHandler: $done->complete(...), + )); + + $srv->addEndpoint(new EndpointConfig('slow', subject: "cancelled-drain.{$id}"), static function (Request $request) use ( + $started, + $release, + ): void { + $started->complete(); + $release->getFuture()->await(); + $request->respond(new Response('done')); + }); + + $reply = async($nc->request(...), "cancelled-drain.{$id}", cancellation: new TimeoutCancellation(5)); + $started->getFuture()->await(new TimeoutCancellation(5)); + + try { + $srv->drain(new TimeoutCancellation(0.2)); + self::fail('The handler is blocked, the drain must be cancelled.'); + } catch (CancelledException) { + } + + // Only the wait is cancelled: the service is still draining. + self::assertSame(Service\State::Stopping, $srv->state); + self::assertFalse($done->getFuture()->isComplete()); + + $release->complete(); + + self::assertSame('done', $reply->await()->message->payload); + self::assertSame($srv, $done->getFuture()->await(new TimeoutCancellation(5))); + self::assertSame(Service\State::Stopped, $srv->state); + + foreach (["cancelled-drain.{$id}", "\$SRV.PING.CancelledDrainService.{$srv->info()->identity->id}"] as $subject) { + try { + $nc->request($subject, cancellation: new TimeoutCancellation(5)); + self::fail("Nothing must be subscribed to {$subject} after the service has stopped."); + } catch (RequestHasNoResponders) { + } + } + } + + public function testUnencodableJsonResponseIsRepliedWithError(): void + { + $nc = $this->client(); + $id = Id\generateUniqueId(10); + + $srv = $nc->createService(new ServiceConfig( + name: 'JsonErrorService', + version: '1.0.0', + )); + + $srv->addEndpoint(new EndpointConfig('nan', subject: "json-error.{$id}"), static function (Request $request): void { + $request->respondJson(NAN); + }); + + $response = $nc->request("json-error.{$id}", cancellation: new TimeoutCancellation(5)); + + self::assertSame(500, $response->message->headers?->get(Header\ServiceErrorCode::header())); + + $stats = $srv->stats()->endpoints; + self::assertCount(1, $stats); + self::assertSame(1, $stats[0]->numErrors); + self::assertInstanceOf(\JsonException::class, $stats[0]->lastError?->getPrevious()); + + $srv->stop(); + } + + public function testEndpointsCannotBeAddedOnceStopping(): void + { + $nc = $this->client(); + $id = Id\generateUniqueId(10); + + /** @var DeferredFuture $started */ + $started = new DeferredFuture(); + + /** @var DeferredFuture $release */ + $release = new DeferredFuture(); + + $srv = $nc->createService(new ServiceConfig( + name: 'StoppingService', + version: '1.0.0', + )); + + $group = $srv->addGroup(new GroupConfig("stopping.{$id}")); + + $srv->addEndpoint(new EndpointConfig('slow', subject: "stopping.{$id}.slow"), static function (Request $request) use ( + $started, + $release, + ): void { + $started->complete(); + $release->getFuture()->await(); + $request->respond(new Response('done')); + }); + + $reply = async($nc->request(...), "stopping.{$id}.slow", cancellation: new TimeoutCancellation(5)); + $started->getFuture()->await(new TimeoutCancellation(5)); + + $drain = async($srv->drain(...)); + delay(0.1); + self::assertSame(Service\State::Stopping, $srv->state); + + $echo = static function (Request $request): void { + $request->respond(new Response($request->data)); + }; + + $adds = [ + 'service while draining' => static fn() => $srv->addEndpoint(new EndpointConfig('late', subject: "stopping.{$id}.late"), $echo), + 'group while draining' => static fn() => $group->addEndpoint(new EndpointConfig('grouped'), $echo), + 'new group while draining' => static fn() => $srv->addGroup(new GroupConfig('late')), + 'nested group while draining' => static fn() => $group->addGroup(new GroupConfig('nested')), + ]; + + foreach ($adds as $case => $add) { + try { + $add(); + self::fail("An endpoint must not be added to a stopping service: {$case}."); + } catch (ServiceAlreadyStopped) { + } + } + + $release->complete(); + self::assertSame('done', $reply->await()->message->payload); + $drain->await(new TimeoutCancellation(5)); + self::assertSame(Service\State::Stopped, $srv->state); + + try { + $srv->addEndpoint(new EndpointConfig('after', subject: "stopping.{$id}.after"), $echo); + self::fail('An endpoint must not be added to a stopped service.'); + } catch (ServiceAlreadyStopped) { + } + + foreach (["stopping.{$id}.late", "stopping.{$id}.grouped", "stopping.{$id}.after"] as $subject) { + try { + $nc->request($subject, cancellation: new TimeoutCancellation(5)); + self::fail("Nothing must be subscribed to {$subject}."); + } catch (RequestHasNoResponders) { + } + } + + self::assertSame([], $srv->info()->endpoints); + } + + public function testDoneHandlerIsCalledOnce(): void + { + /** @var \ArrayObject $done */ + $done = new \ArrayObject(); + + $srv = $this->client()->createService(new ServiceConfig( + name: 'DoneService', + version: '1.0.0', + doneHandler: static function (Service $srv) use ($done): void { + $done[] = $srv; + }, + )); + + $srv->addEndpoint(new EndpointConfig('echo', subject: 'done.' . Id\generateUniqueId(10)), static fn() => null); + + self::assertSame(Service\State::Alive, $srv->state); + + $srv->stop(); + $srv->stop(); + $srv->drain(); + + delay(0.05); + + self::assertSame(Service\State::Stopped, $srv->state); + self::assertCount(1, $done); + self::assertSame($srv, $done[0]); + self::assertSame([], $srv->info()->endpoints); + } + + public function testDoneHandlerFailureIsReportedAfterShutdown(): void + { + /** @var DeferredFuture $reported */ + $reported = new DeferredFuture(); + $failure = new \RuntimeException('Done handler failed'); + $nc = $this->client(); + + $srv = $nc->createService(new ServiceConfig( + name: 'DoneFailureService', + version: '1.0.0', + doneHandler: static function (Service $srv) use ($failure): void { + $srv->stop(); + + throw $failure; + }, + errorHandler: static function (Service $srv, \Throwable $e) use ($reported): void { + $srv->drain(); + $reported->complete([$srv, $e]); + }, + )); + + $subject = 'done-failure.' . Id\generateUniqueId(10); + $srv->addEndpoint(new EndpointConfig('echo', subject: $subject), static fn() => null); + $srv->stop(); + + self::assertSame([$srv, $failure], $reported->getFuture()->await(new TimeoutCancellation(5))); + self::assertSame(Service\State::Stopped, $srv->state); + self::assertSame([], $srv->info()->endpoints); + + $this->expectException(RequestHasNoResponders::class); + $nc->request($subject, cancellation: new TimeoutCancellation(5)); + } + + public function testCallbackFailuresReachEventLoopErrorHandler(): void + { + $nc = $this->client(); + $previousHandler = EventLoop::getErrorHandler(); + + try { + foreach ([false, true] as $withErrorHandler) { + /** @var DeferredFuture<\Throwable> $reported */ + $reported = new DeferredFuture(); + EventLoop::setErrorHandler($reported->complete(...)); + + $doneFailure = new \RuntimeException('Done handler failed'); + $errorFailure = new \RuntimeException('Error handler failed'); + $srv = $nc->createService(new ServiceConfig( + name: 'CallbackFailureService', + version: '1.0.0', + doneHandler: static function (Service $_) use ($doneFailure): void { + throw $doneFailure; + }, + errorHandler: $withErrorHandler ? static function (Service $_, \Throwable $e) use ($doneFailure, $errorFailure): void { + self::assertSame($doneFailure, $e); + + throw $errorFailure; + } : null, + )); + + $srv->addEndpoint(new EndpointConfig('echo', subject: 'callback-failure.' . Id\generateUniqueId(10)), static fn() => null); + $srv->stop(); + + self::assertSame( + $withErrorHandler ? $errorFailure : $doneFailure, + $reported->getFuture()->await(new TimeoutCancellation(5)), + ); + self::assertSame(Service\State::Stopped, $srv->state); + } + } finally { + EventLoop::setErrorHandler($previousHandler); + } + } + + public function testServiceStopsWithClient(): void + { + /** @var DeferredFuture $done */ + $done = new DeferredFuture(); + /** @var \ArrayObject $errors */ + $errors = new \ArrayObject(); + + $nc = $this->client(); + + $srv = $nc->createService(new ServiceConfig( + name: 'ClientStopService', + version: '1.0.0', + doneHandler: $done->complete(...), + errorHandler: static function (Service $_, \Throwable $e) use ($errors): void { + $errors[] = $e; + }, + )); + + $srv->addEndpoint(new EndpointConfig('echo', subject: 'client-stop.' . Id\generateUniqueId(10)), static fn() => null); + + $nc->stop(); + + self::assertSame($srv, $done->getFuture()->await(new TimeoutCancellation(5))); + self::assertSame(Service\State::Stopped, $srv->state); + self::assertCount(0, $errors); + } + + public function testErrorHandlerIsCalledWhenSubscriptionFails(): void + { + $nc = $this->clientWithConfig(new Config( + urls: [$this->natsAddress()], + user: 'restricted', + password: 'Pswd1', + )); + + /** @var DeferredFuture $done */ + $done = new DeferredFuture(); + /** @var \ArrayObject $errors */ + $errors = new \ArrayObject(); + + // The user may subscribe to allowed.> only: the $SRV control subjects are rejected by the server. + $srv = $nc->createService(new ServiceConfig( + name: 'ForbiddenService', + version: '1.0.0', + doneHandler: $done->complete(...), + errorHandler: static function (Service $_, \Throwable $e) use ($errors): void { + $errors[] = $e; + }, + )); + + $srv->addEndpoint(new EndpointConfig('echo', subject: 'allowed.' . Id\generateUniqueId(10)), static fn() => null); + + self::assertSame($srv, $done->getFuture()->await(new TimeoutCancellation(5))); + self::assertSame(Service\State::Stopped, $srv->state); + + // All the $SRV subscriptions are rejected, but only the first failure is reported: the service is stopping then. + delay(0.05); + self::assertCount(1, $errors); + self::assertInstanceOf(PermissionViolated::class, $errors[0]); + } + public function testGroups(): void { $nc = $this->client();