fix: tighten micro service semantics

README.md

@@ -56,7 +56,7 @@ For full runnable scripts grouped by feature, see the example files below. That
 | **JetStream Publish** | Synchronous publish acknowledgements, deduplication (msgId), per-message TTL, optimistic concurrency, retry, stallWait |
 | **Key-Value Store** | Put, get, getRevision, create (CAS + TTL), update (CAS), delete, purge (+ TTL), purgeDeletes, history, watch, watchFiltered, listKeys, listKeysFiltered, bucket CRUD |
 | **Object Store** | Put (bytes/string/file), get, getInfo, delete, list (+ showDeleted), addLink, addBucketLink, seal, watch, chunked transfer, metadata, compression |
-| **Micro Services** | Endpoints, groups, discovery ($SRV.PING/INFO/STATS), statistics, pending limits, handler classes |
+| **Micro Services** | Endpoints, groups, discovery ($SRV.PING/INFO/STATS), statistics, pending limits, graceful stop, runtime error visibility, handler classes |
 
 ## Architecture
 
@@ -496,6 +496,12 @@ $math->addEndpoint('sqrt', function (RequestInterface $req) {
 // Auto-discovery: $SRV.PING, $SRV.INFO.calculator, $SRV.STATS.calculator
 $info = $svc->info();
 $stats = $svc->stats();
+
+// reset() starts a new stats window; lastError() exposes the latest runtime failure
+$svc->reset();
+$lastError = $svc->lastError();
+
+// stop() drains service subscriptions before shutdown
 $svc->stop();
 ```
 

docs/examples/micro/01-service.php

@@ -1,10 +1,11 @@
-<?php
+<?php declare(strict_types=1);
 
 /**
  * Micro Service API - microservices framework built on NATS.
  *
  * Automatically registers discovery subjects ($SRV.PING/INFO/STATS),
- * supports endpoint grouping, queue groups, statistics.
+ * supports endpoint grouping, queue groups, statistics, runtime error visibility,
+ * and graceful service shutdown.
  */
 
 require_once __DIR__ . '/../../../vendor/autoload.php';
@@ -27,17 +28,17 @@
     description: 'Calculator microservice',
     metadata: ['author' => 'team-php'],
     queueGroup: 'q',  // Load balancing
-    errorHandler: function ($service, \Throwable $e) {
+    errorHandler: function ($service, \Throwable $e): void {
         echo "[SVC ERROR] {$e->getMessage()}\n";
     },
-    doneHandler: function ($service) {
+    doneHandler: function ($service): void {
         echo "[SVC] Service stopped.\n";
     },
 ));
 
 // --- Adding endpoints (Closure) ---
 
-$svc->addEndpoint('add', function (RequestInterface $req) {
+$svc->addEndpoint('add', function (RequestInterface $req): void {
     $data = json_decode($req->data(), true);
     $result = ($data['a'] ?? 0) + ($data['b'] ?? 0);
     $req->respondJson(['result' => $result]);
@@ -46,7 +47,7 @@
     metadata: ['description' => 'Adds two numbers'],
 ));
 
-$svc->addEndpoint('multiply', function (RequestInterface $req) {
+$svc->addEndpoint('multiply', function (RequestInterface $req): void {
     $data = json_decode($req->data(), true);
     $result = ($data['a'] ?? 0) * ($data['b'] ?? 0);
     $req->respondJson(['result' => $result]);
@@ -78,15 +79,15 @@ public function handle(RequestInterface $request): void
 // --- Grouping endpoints ---
 
 $mathGroup = $svc->addGroup('math');
-$mathGroup->addEndpoint('sqrt', function (RequestInterface $req) {
+$mathGroup->addEndpoint('sqrt', function (RequestInterface $req): void {
     $data = json_decode($req->data(), true);
     $req->respondJson(['result' => sqrt($data['n'] ?? 0)]);
 });
 // Subject will be: math.sqrt
 
 // Nested groups
 $advancedGroup = $mathGroup->addGroup('advanced');
-$advancedGroup->addEndpoint('factorial', function (RequestInterface $req) {
+$advancedGroup->addEndpoint('factorial', function (RequestInterface $req): void {
     $data = json_decode($req->data(), true);
     $n = $data['n'] ?? 0;
     $result = 1;
@@ -135,10 +136,14 @@ public function handle(RequestInterface $request): void
 }
 
 // --- Reset statistics ---
+// reset() starts a new stats window and clears the current lastError()
 
 $svc->reset();
+$lastError = $svc->lastError();
+echo "Last error: " . ($lastError?->getMessage() ?? 'none') . "\n";
 
 // --- Graceful stop ---
+// stop() drains service subscriptions before shutdown.
 
 $svc->stop();
 echo "Stopped: " . ($svc->isStopped() ? 'yes' : 'no') . "\n";

src/Message.php

@@ -12,6 +12,8 @@ public function __construct(
         public ?string $replyTo = null,
         public ?Headers $headers = null,
         private ?Connection $connection = null,
+        private ?\Closure $beforeRespond = null,
+        private ?\Closure $afterRespond = null,
     ) {}
 
     public function size(): int
@@ -31,13 +33,20 @@ public function respond(string $data, ?Headers $headers = null): void
         if ($this->connection === null) {
             throw new NatsException('No connection available for respond');
         }
+        if ($this->beforeRespond !== null) {
+            ($this->beforeRespond)();
+        }
 
         $msg = new self(
             subject: $this->replyTo,
             data: $data,
             headers: $headers,
         );
         $this->connection->publishMessage($msg);
+
+        if ($this->afterRespond !== null) {
+            ($this->afterRespond)();
+        }
     }
 
     public function hasHeaders(): bool

src/Micro/Group.php

@@ -33,6 +33,8 @@ public function addEndpoint(
             metadata: $config->metadata,
             queueGroup: $queueGroup,
             queueGroupDisabled: $queueGroupDisabled,
+            pendingMsgLimit: $config->pendingMsgLimit,
+            pendingBytesLimit: $config->pendingBytesLimit,
         );
 
         $this->service->addEndpoint($name, $handler, $endpointConfig);

src/Micro/Request.php

@@ -11,23 +11,27 @@
 
 final class Request implements RequestInterface
 {
-    public function __construct(
-        private readonly Message $msg,
-        private readonly Connection $connection,
-    ) {}
+    private bool $responded = false;
+    private readonly Message $msg;
+    private readonly Connection $connection;
 
-    public function respond(string $data, ?Headers $headers = null): void
+    public function __construct(Message $msg, Connection $connection)
     {
-        if ($this->msg->replyTo === null) {
-            throw new NatsException('No reply subject');
-        }
-
-        $response = new Message(
-            subject: $this->msg->replyTo,
-            data: $data,
-            headers: $headers,
+        $this->connection = $connection;
+        $this->msg = new Message(
+            subject: $msg->subject,
+            data: $msg->data,
+            replyTo: $msg->replyTo,
+            headers: $msg->headers,
+            connection: $connection,
+            beforeRespond: $this->assertCanRespond(...),
+            afterRespond: $this->markResponded(...),
         );
-        $this->connection->publishMessage($response);
+    }
+
+    public function respond(string $data, ?Headers $headers = null): void
+    {
+        $this->sendResponse($data, $headers);
     }
 
     public function respondJson(mixed $data, ?Headers $headers = null): void
@@ -39,20 +43,16 @@ public function respondJson(mixed $data, ?Headers $headers = null): void
 
     public function error(string $code, string $description, string $data = ''): void
     {
-        if ($this->msg->replyTo === null) {
-            throw new NatsException('No reply subject');
-        }
-
         $headers = new Headers();
         $headers->set('Nats-Service-Error', $description);
         $headers->set('Nats-Service-Error-Code', $code);
 
-        $response = new Message(
-            subject: $this->msg->replyTo,
-            data: $data,
-            headers: $headers,
-        );
-        $this->connection->publishMessage($response);
+        $this->sendResponse($data, $headers);
+    }
+
+    public function hasResponded(): bool
+    {
+        return $this->responded;
     }
 
     public function data(): string
@@ -79,4 +79,33 @@ public function msg(): Message
     {
         return $this->msg;
     }
+
+    private function sendResponse(string $data, ?Headers $headers = null): void
+    {
+        if ($this->msg->replyTo === null) {
+            throw new NatsException('No reply subject');
+        }
+
+        $this->assertCanRespond();
+
+        $response = new Message(
+            subject: $this->msg->replyTo,
+            data: $data,
+            headers: $headers,
+        );
+        $this->connection->publishMessage($response);
+        $this->markResponded();
+    }
+
+    private function assertCanRespond(): void
+    {
+        if ($this->responded) {
+            throw new NatsException('Response already sent');
+        }
+    }
+
+    private function markResponded(): void
+    {
+        $this->responded = true;
+    }
 }

src/Micro/RequestInterface.php

@@ -15,6 +15,8 @@ public function respondJson(mixed $data, ?Headers $headers = null): void;
 
     public function error(string $code, string $description, string $data = ''): void;
 
+    public function hasResponded(): bool;
+
     public function data(): string;
 
     public function headers(): ?Headers;