Add micro service errors, handlers and drain

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
+<?php
+
+declare(strict_types=1);
+
+use Thesis\Nats;
+use Thesis\Nats\Micro;
+
+require __DIR__ . '/vendor/autoload.php';
+
+$nc = new Nats\Client(
+    Nats\Config::default(),
+);
+
+$srv = $nc->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:<message>`.
+
+```php
+<?php
+
+declare(strict_types=1);
+
+use Thesis\Nats;
+use Thesis\Nats\Micro;
+
+require __DIR__ . '/vendor/autoload.php';
+
+$nc = new Nats\Client(
+    Nats\Config::default(),
+);
+
+$srv = $nc->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
+<?php
+
+declare(strict_types=1);
+
+use Thesis\Nats;
+use Thesis\Nats\Micro;
+
+require __DIR__ . '/vendor/autoload.php';
+
+$nc = new Nats\Client(
+    Nats\Config::default(),
+);
+
+$srv = $nc->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
+<?php
+
+declare(strict_types=1);
+
+use Thesis\Nats;
+use Thesis\Nats\Micro;
+
+require __DIR__ . '/vendor/autoload.php';
+
+$nc = new Nats\Client(
+    Nats\Config::default(),
+);
+
+$srv = $nc->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.

src/Exception/ServiceAlreadyStopped.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Nats\Exception;
+
+use Thesis\Nats\NatsException;
+
+/**
+ * @api
+ */
+final class ServiceAlreadyStopped extends NatsException {}

src/Micro/EndpointStats.php

@@ -7,7 +7,7 @@
 /**
  * @api
  */
-final class EndpointStats
+final readonly class EndpointStats
 {
     /**
      * @param non-empty-string $name
@@ -17,6 +17,7 @@ final class EndpointStats
      * @param non-negative-int $numErrors
      * @param non-negative-int $processingTime total processing time in nanoseconds
      * @param non-negative-int $averageProcessingTime average processing time in nanoseconds
+     * @param mixed $data the value returned by {@see ServiceConfig::$statsHandler}, reported as JSON
      */
     public function __construct(
         public string $name,
@@ -27,6 +28,6 @@ public function __construct(
         public ?\Throwable $lastError = null,
         public int $processingTime = 0,
         public int $averageProcessingTime = 0,
-        public ?string $data = null,
+        public mixed $data = null,
     ) {}
 }

src/Micro/Group.php

@@ -5,6 +5,7 @@
 namespace Thesis\Nats\Micro;
 
 use Amp\Cancellation;
+use Thesis\Nats\Exception\ServiceAlreadyStopped;
 use Thesis\Nats\NatsException;
 
 /**
@@ -25,6 +26,7 @@ public function __construct(
 
     /**
      * @param callable(Request): void $handler
+     * @throws ServiceAlreadyStopped once the service has started stopping
      * @throws NatsException
      */
     public function addEndpoint(
@@ -46,6 +48,9 @@ public function addEndpoint(
         return $this;
     }
 
+    /**
+     * @throws ServiceAlreadyStopped once the service has started stopping
+     */
     public function addGroup(GroupConfig $config): self
     {
         return $this->svc->addGroup(

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;
+    }
 }

src/Micro/Internal/RequestOutcome.php

@@ -0,0 +1,25 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Nats\Micro\Internal;
+
+/**
+ * @internal
+ */
+final class RequestOutcome
+{
+    public private(set) bool $replied = false;
+
+    public private(set) ?\Throwable $error = null;
+
+    public function markReplied(): void
+    {
+        $this->replied = true;
+    }
+
+    public function fail(\Throwable $error): void
+    {
+        $this->error = $error;
+    }
+}

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();
+    }
 }