From fd0699a7d081713e88452d2a24ea16cd404c5de3 Mon Sep 17 00:00:00 2001 From: Your Name Date: Mon, 30 Mar 2026 22:37:01 +0200 Subject: [PATCH 1/5] fix: tighten micro service semantics --- README.md | 8 +- docs/examples/micro/01-service.php | 21 ++-- src/Micro/Group.php | 2 + src/Micro/Request.php | 48 +++++--- src/Micro/RequestInterface.php | 2 + src/Micro/Service.php | 155 ++++++++++++++++++------ src/Micro/ServiceConfig.php | 1 - src/Micro/ServiceInterface.php | 2 + tests/Integration/Micro/ServiceTest.php | 119 ++++++++++++++++++ 9 files changed, 290 insertions(+), 68 deletions(-) diff --git a/README.md b/README.md index cb3d6c3..051c5bf 100644 --- a/README.md +++ b/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(); ``` diff --git a/docs/examples/micro/01-service.php b/docs/examples/micro/01-service.php index 0525b43..f3434e0 100644 --- a/docs/examples/micro/01-service.php +++ b/docs/examples/micro/01-service.php @@ -1,10 +1,11 @@ - '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,7 +79,7 @@ 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)]); }); @@ -86,7 +87,7 @@ public function handle(RequestInterface $request): void // 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"; diff --git a/src/Micro/Group.php b/src/Micro/Group.php index 904621b..aa5cf57 100644 --- a/src/Micro/Group.php +++ b/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); diff --git a/src/Micro/Request.php b/src/Micro/Request.php index 14e7c90..97300b9 100644 --- a/src/Micro/Request.php +++ b/src/Micro/Request.php @@ -11,6 +11,8 @@ final class Request implements RequestInterface { + private bool $responded = false; + public function __construct( private readonly Message $msg, private readonly Connection $connection, @@ -18,16 +20,7 @@ public function __construct( public function respond(string $data, ?Headers $headers = null): void { - if ($this->msg->replyTo === null) { - throw new NatsException('No reply subject'); - } - - $response = new Message( - subject: $this->msg->replyTo, - data: $data, - headers: $headers, - ); - $this->connection->publishMessage($response); + $this->sendResponse($data, $headers); } public function respondJson(mixed $data, ?Headers $headers = null): void @@ -39,20 +32,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 +68,23 @@ 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'); + } + + if ($this->responded) { + throw new NatsException('Response already sent'); + } + + $response = new Message( + subject: $this->msg->replyTo, + data: $data, + headers: $headers, + ); + $this->connection->publishMessage($response); + $this->responded = true; + } } diff --git a/src/Micro/RequestInterface.php b/src/Micro/RequestInterface.php index d760f76..50cf03d 100644 --- a/src/Micro/RequestInterface.php +++ b/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; diff --git a/src/Micro/Service.php b/src/Micro/Service.php index ffaa03d..0a002c8 100644 --- a/src/Micro/Service.php +++ b/src/Micro/Service.php @@ -8,6 +8,7 @@ use Sofascore\Nats\Enum\ServiceVerb; use Sofascore\Nats\Inbox; use Sofascore\Nats\Message; +use Sofascore\Nats\NatsException; use Sofascore\Nats\Subscription; final class Service implements ServiceInterface @@ -34,8 +35,10 @@ public static function controlSubject(ServiceVerb $verb, string $name = '', stri } private readonly string $id; - private readonly \DateTimeImmutable $started; + private \DateTimeImmutable $started; private bool $stopped = false; + private ?\Throwable $lastError = null; + private int $handlerDepth = 0; /** @var array */ private array $endpoints = []; @@ -55,12 +58,6 @@ public static function create(Connection $conn, ServiceConfig $config): self { $service = new self($conn, $config); $service->setupControlSubjects(); - - // Add default endpoint if configured - if ($config->endpoint !== null) { - // Requires a handler — skip if not provided - } - return $service; } @@ -69,20 +66,18 @@ public function addEndpoint( \Closure|HandlerInterface $handler, ?EndpointConfig $config = null, ): self { - $config ??= new EndpointConfig(); - $subject = $config->subject ?? $name; + if ($this->stopped) { + throw new NatsException('Service is stopped'); + } - $queueGroup = match (true) { - $config->queueGroupDisabled => null, - $config->queueGroup !== null => $config->queueGroup, - $this->config->queueGroupDisabled => null, - $this->config->queueGroup !== null => $this->config->queueGroup, - default => 'q', - }; + $config ??= new EndpointConfig(); + $subject = $this->subjectFor($name, $config); + $queueGroup = $this->queueGroupFor($config); $wrappedHandler = function (Message $msg) use ($name, $handler): void { $request = new Request($msg, $this->connection); $startTime = hrtime(true); + $this->handlerDepth++; try { if ($handler instanceof HandlerInterface) { @@ -105,17 +100,18 @@ public function addEndpoint( $stats['last_error'] = $e->getMessage(); $this->endpoints[$name]['stats'] = $stats; - $errorHandler = $this->config->errorHandler; - if ($errorHandler !== null) { - $errorHandler($this, $e); - } + $this->recordError($e); // Send error response if not already responded - try { - $request->error('500', $e->getMessage()); - } catch (\Throwable) { - // Ignore - response may already be sent + if (!$request->hasResponded()) { + try { + $request->error('500', $e->getMessage()); + } catch (\Throwable $sendError) { + $this->recordError($sendError); + } } + } finally { + $this->handlerDepth--; } }; @@ -148,6 +144,10 @@ public function addEndpoint( public function addGroup(string $name, ?string $queueGroup = null, bool $queueGroupDisabled = false): GroupInterface { + if ($this->stopped) { + throw new NatsException('Service is stopped'); + } + return new Group($this, $name, $queueGroup, $queueGroupDisabled); } @@ -156,7 +156,7 @@ public function info(): ServiceInfo /** @var list}> $endpoints */ $endpoints = []; foreach ($this->endpoints as $name => $ep) { - $subject = $ep['config']->subject ?? $name; + $subject = $this->subjectFor($name, $ep['config']); if ($ep['config']->metadata !== []) { $endpoints[] = [ 'name' => $name, @@ -193,7 +193,7 @@ public function stats(): ServiceStats $epStats = new EndpointStats( name: $name, - subject: $ep['config']->subject ?? $name, + subject: $this->subjectFor($name, $ep['config']), numRequests: $totalReqs, numErrors: $ep['stats']['errors'], processingTime: $totalTime, @@ -242,6 +242,9 @@ public function reset(): void 'last_error' => null, ]; } + + $this->lastError = null; + $this->started = new \DateTimeImmutable(); } public function stop(): void @@ -251,15 +254,33 @@ public function stop(): void } $this->stopped = true; + $subs = $this->serviceSubscriptions(); - // Unsubscribe all endpoints - foreach ($this->endpoints as $ep) { - $ep['sub']->unsubscribe(); - } + if ($this->connection->isConnected()) { + foreach ($subs as $sub) { + $sub->drain(); + } + + if ($this->handlerDepth === 0) { + try { + $this->connection->flush(0.5); + } catch (\Throwable $e) { + $this->recordError($e); + foreach ($subs as $sub) { + $sub->unsubscribe(); + } + } + } + } else { + foreach ($subs as $sub) { + $sub->unsubscribe(); + } - // Unsubscribe control subjects - foreach ($this->controlSubs as $sub) { - $sub->unsubscribe(); + if (!$this->connection->isClosed()) { + $this->recordError(new NatsException( + 'Service stopped without graceful drain because the connection was not connected', + )); + } } $doneHandler = $this->config->doneHandler; @@ -273,6 +294,11 @@ public function isStopped(): bool return $this->stopped; } + public function lastError(): ?\Throwable + { + return $this->lastError; + } + public function id(): string { return $this->id; @@ -323,18 +349,71 @@ private function handlePing(Message $msg): void $response['metadata'] = $this->config->metadata; } - $msg->respond(json_encode($response, JSON_THROW_ON_ERROR)); + $this->replyControl($msg, $response); } private function handleInfo(Message $msg): void { - $info = $this->info(); - $msg->respond(json_encode($info->toArray(), JSON_THROW_ON_ERROR)); + $this->replyControl($msg, $this->info()->toArray()); } private function handleStats(Message $msg): void { - $stats = $this->stats(); - $msg->respond(json_encode($stats->toArray(), JSON_THROW_ON_ERROR)); + $this->replyControl($msg, $this->stats()->toArray()); + } + + /** + * @return list + */ + private function serviceSubscriptions(): array + { + $subs = []; + foreach ($this->endpoints as $ep) { + $subs[] = $ep['sub']; + } + + return array_merge($subs, $this->controlSubs); + } + + private function subjectFor(string $name, EndpointConfig $config): string + { + return $config->subject ?? $name; + } + + private function queueGroupFor(EndpointConfig $config): ?string + { + return match (true) { + $config->queueGroupDisabled => null, + $config->queueGroup !== null => $config->queueGroup, + $this->config->queueGroupDisabled => null, + $this->config->queueGroup !== null => $this->config->queueGroup, + default => 'q', + }; + } + + /** + * @param array $payload + */ + private function replyControl(Message $msg, array $payload): void + { + $this->handlerDepth++; + + try { + $msg->respond(json_encode($payload, JSON_THROW_ON_ERROR)); + } catch (\Throwable $e) { + $this->recordError($e); + } finally { + $this->handlerDepth--; + } + } + + private function recordError(\Throwable $e): void + { + $this->lastError = $e; + + $errorHandler = $this->config->errorHandler; + if ($errorHandler !== null) { + $errorHandler($this, $e); + } } } diff --git a/src/Micro/ServiceConfig.php b/src/Micro/ServiceConfig.php index fdfe9f9..0b44194 100644 --- a/src/Micro/ServiceConfig.php +++ b/src/Micro/ServiceConfig.php @@ -12,7 +12,6 @@ public function __construct( public ?string $description = null, /** @var array */ public array $metadata = [], - public ?EndpointConfig $endpoint = null, public ?string $queueGroup = null, public bool $queueGroupDisabled = false, public ?\Closure $statsHandler = null, diff --git a/src/Micro/ServiceInterface.php b/src/Micro/ServiceInterface.php index 7540dd0..b0463c6 100644 --- a/src/Micro/ServiceInterface.php +++ b/src/Micro/ServiceInterface.php @@ -24,6 +24,8 @@ public function stop(): void; public function isStopped(): bool; + public function lastError(): ?\Throwable; + public function id(): string; public function name(): string; diff --git a/tests/Integration/Micro/ServiceTest.php b/tests/Integration/Micro/ServiceTest.php index c1aa8f6..e4bfc91 100644 --- a/tests/Integration/Micro/ServiceTest.php +++ b/tests/Integration/Micro/ServiceTest.php @@ -10,6 +10,7 @@ use Sofascore\Nats\Micro\RequestInterface; use Sofascore\Nats\Micro\Service; use Sofascore\Nats\Micro\ServiceConfig; +use Sofascore\Nats\NatsException; use Sofascore\Nats\Tests\Integration\NatsUrlTrait; use PHPUnit\Framework\TestCase; @@ -256,6 +257,34 @@ public function testServiceReset(): void $stats = $svc->stats(); self::assertSame(0, $stats->endpoints[0]->numRequests); + self::assertNull($svc->lastError()); + + $svc->stop(); + } + + public function testServiceResetStartsNewStatsWindow(): void + { + $svc = Service::create($this->conn, new ServiceConfig( + name: 'reset-window-svc', + version: '1.0.0', + )); + + $svc->addEndpoint('fail', function (): void { + throw new \RuntimeException('boom'); + }, new EndpointConfig(subject: 'svc.reset.window')); + + $reply = $this->conn->request('svc.reset.window', '', 2.0); + self::assertSame('boom', $reply->headers?->get('Nats-Service-Error')); + self::assertSame('boom', $svc->lastError()?->getMessage()); + + $startedBeforeReset = $svc->stats()->started->format('U.u'); + usleep(10_000); + + $svc->reset(); + + $startedAfterReset = $svc->stats()->started->format('U.u'); + self::assertNotSame($startedBeforeReset, $startedAfterReset); + self::assertNull($svc->lastError()); $svc->stop(); } @@ -306,4 +335,94 @@ public function testStopIsIdempotent(): void $svc->stop(); // Should not throw self::assertTrue($svc->isStopped()); } + + public function testHandlerFailureIsExposedThroughLastErrorAndStats(): void + { + $svc = Service::create($this->conn, new ServiceConfig( + name: 'last-error-svc', + version: '1.0.0', + )); + + $svc->addEndpoint('explode', function (): void { + throw new \RuntimeException('handler exploded'); + }, new EndpointConfig(subject: 'svc.last.error')); + + $reply = $this->conn->request('svc.last.error', '', 2.0); + + self::assertSame('handler exploded', $reply->headers?->get('Nats-Service-Error')); + self::assertInstanceOf(\RuntimeException::class, $svc->lastError()); + self::assertSame('handler exploded', $svc->lastError()?->getMessage()); + + $stats = $svc->stats(); + self::assertSame(1, $stats->endpoints[0]->numRequests); + self::assertSame(1, $stats->endpoints[0]->numErrors); + self::assertSame('handler exploded', $stats->endpoints[0]->lastError); + + $svc->stop(); + } + + public function testRequestHasRespondedState(): void + { + $states = []; + + $svc = Service::create($this->conn, new ServiceConfig( + name: 'responded-svc', + version: '1.0.0', + )); + + $svc->addEndpoint('echo', function (RequestInterface $req) use (&$states): void { + $states[] = $req->hasResponded(); + $req->respond('ok'); + $states[] = $req->hasResponded(); + }, new EndpointConfig(subject: 'svc.responded.state')); + + $reply = $this->conn->request('svc.responded.state', '', 2.0); + + self::assertSame('ok', $reply->data); + self::assertSame([false, true], $states); + + $svc->stop(); + } + + public function testResponseSendFailureIsObservable(): void + { + $svc = Service::create($this->conn, new ServiceConfig( + name: 'response-failure-svc', + version: '1.0.0', + )); + + $svc->addEndpoint('reply', function (RequestInterface $req): void { + $req->respond('ok'); + }, new EndpointConfig(subject: 'svc.response.failure')); + + $this->conn->publish('svc.response.failure', 'fire-and-forget'); + $this->conn->flush(); + $this->conn->process(0.2); + + self::assertSame('No reply subject', $svc->lastError()?->getMessage()); + + $stats = $svc->stats(); + self::assertSame(1, $stats->endpoints[0]->numRequests); + self::assertSame(1, $stats->endpoints[0]->numErrors); + self::assertSame('No reply subject', $stats->endpoints[0]->lastError); + + $svc->stop(); + } + + public function testCannotAddEndpointAfterStop(): void + { + $svc = Service::create($this->conn, new ServiceConfig( + name: 'stopped-svc', + version: '1.0.0', + )); + + $svc->stop(); + + $this->expectException(NatsException::class); + $this->expectExceptionMessage('Service is stopped'); + + $svc->addEndpoint('echo', function (RequestInterface $req): void { + $req->respond('ok'); + }); + } } From 7f3887ad986f28fe17f460065eca51bedc57b41a Mon Sep 17 00:00:00 2001 From: Your Name Date: Mon, 30 Mar 2026 23:06:52 +0200 Subject: [PATCH 2/5] fix: clean up micro subscriptions on stop --- src/Micro/Service.php | 32 +++++++++++++++++++++++++ tests/Integration/Micro/ServiceTest.php | 18 ++++++++++++++ 2 files changed, 50 insertions(+) diff --git a/src/Micro/Service.php b/src/Micro/Service.php index 0a002c8..3cb106d 100644 --- a/src/Micro/Service.php +++ b/src/Micro/Service.php @@ -39,6 +39,7 @@ public static function controlSubject(ServiceVerb $verb, string $name = '', stri private bool $stopped = false; private ?\Throwable $lastError = null; private int $handlerDepth = 0; + private bool $pendingStopCleanup = false; /** @var array */ private array $endpoints = []; @@ -112,6 +113,7 @@ public function addEndpoint( } } finally { $this->handlerDepth--; + $this->finishPendingStopCleanup(); } }; @@ -264,12 +266,15 @@ public function stop(): void if ($this->handlerDepth === 0) { try { $this->connection->flush(0.5); + $this->finalizeStoppedSubscriptions($subs); } catch (\Throwable $e) { $this->recordError($e); foreach ($subs as $sub) { $sub->unsubscribe(); } } + } else { + $this->pendingStopCleanup = true; } } else { foreach ($subs as $sub) { @@ -404,6 +409,7 @@ private function replyControl(Message $msg, array $payload): void $this->recordError($e); } finally { $this->handlerDepth--; + $this->finishPendingStopCleanup(); } } @@ -416,4 +422,30 @@ private function recordError(\Throwable $e): void $errorHandler($this, $e); } } + + private function finishPendingStopCleanup(): void + { + if (!$this->pendingStopCleanup || $this->handlerDepth !== 0) { + return; + } + + $this->pendingStopCleanup = false; + $this->finalizeStoppedSubscriptions($this->serviceSubscriptions()); + } + + /** + * @param list $subs + */ + private function finalizeStoppedSubscriptions(array $subs): void + { + foreach ($subs as $sub) { + $sub->checkDrainComplete(); + } + + foreach ($subs as $sub) { + if (!$sub->isDrained() && $sub->isValid()) { + $sub->unsubscribe(); + } + } + } } diff --git a/tests/Integration/Micro/ServiceTest.php b/tests/Integration/Micro/ServiceTest.php index e4bfc91..f884b67 100644 --- a/tests/Integration/Micro/ServiceTest.php +++ b/tests/Integration/Micro/ServiceTest.php @@ -336,6 +336,24 @@ public function testStopIsIdempotent(): void self::assertTrue($svc->isStopped()); } + public function testStopCleansUpServiceSubscriptions(): void + { + $svc = Service::create($this->conn, new ServiceConfig( + name: 'cleanup-svc', + version: '1.0.0', + )); + + $svc->addEndpoint('echo', function (RequestInterface $req): void { + $req->respond('ok'); + }, new EndpointConfig(subject: 'svc.cleanup.echo')); + + self::assertGreaterThan(0, $this->conn->numSubscriptions()); + + $svc->stop(); + + self::assertSame(0, $this->conn->numSubscriptions()); + } + public function testHandlerFailureIsExposedThroughLastErrorAndStats(): void { $svc = Service::create($this->conn, new ServiceConfig( From ef489583e9e3ccec4faff047ee659f8e14189b49 Mon Sep 17 00:00:00 2001 From: Your Name Date: Mon, 30 Mar 2026 23:27:52 +0200 Subject: [PATCH 3/5] fix: flush deferred micro stop cleanup --- src/Micro/Service.php | 16 +++++++++++- tests/Integration/Micro/ServiceTest.php | 33 +++++++++++++++++++++++++ 2 files changed, 48 insertions(+), 1 deletion(-) diff --git a/src/Micro/Service.php b/src/Micro/Service.php index 3cb106d..0e5f9c7 100644 --- a/src/Micro/Service.php +++ b/src/Micro/Service.php @@ -430,7 +430,21 @@ private function finishPendingStopCleanup(): void } $this->pendingStopCleanup = false; - $this->finalizeStoppedSubscriptions($this->serviceSubscriptions()); + $subs = $this->serviceSubscriptions(); + + if ($this->connection->isConnected()) { + try { + $this->connection->flush(0.5); + } catch (\Throwable $e) { + $this->recordError($e); + foreach ($subs as $sub) { + $sub->unsubscribe(); + } + return; + } + } + + $this->finalizeStoppedSubscriptions($subs); } /** diff --git a/tests/Integration/Micro/ServiceTest.php b/tests/Integration/Micro/ServiceTest.php index f884b67..0b89938 100644 --- a/tests/Integration/Micro/ServiceTest.php +++ b/tests/Integration/Micro/ServiceTest.php @@ -354,6 +354,39 @@ public function testStopCleansUpServiceSubscriptions(): void self::assertSame(0, $this->conn->numSubscriptions()); } + public function testStopInsideHandlerFlushesReplyAndCleanup(): void + { + $client = Connection::connect(self::natsUrl()); + + try { + $svc = Service::create($this->conn, new ServiceConfig( + name: 'self-stop-svc', + version: '1.0.0', + )); + + $svc->addEndpoint('bye', function (RequestInterface $req) use ($svc): void { + $req->respond('bye'); + $svc->stop(); + }, new EndpointConfig(subject: 'svc.self.stop')); + + $this->conn->flush(); + + $replyInbox = $client->newInbox(); + $replySub = $client->subscribeSync($replyInbox); + + $client->publish('svc.self.stop', '', $replyInbox); + $client->flush(); + $this->conn->process(0.5); + $reply = $replySub->nextMessage(0.5); + + self::assertSame('bye', $reply->data); + self::assertTrue($svc->isStopped()); + self::assertSame(0, $this->conn->numSubscriptions()); + } finally { + $client->close(); + } + } + public function testHandlerFailureIsExposedThroughLastErrorAndStats(): void { $svc = Service::create($this->conn, new ServiceConfig( From 0756d64f3fa3b7e07283c868200e781474b165fc Mon Sep 17 00:00:00 2001 From: Your Name Date: Mon, 30 Mar 2026 23:37:22 +0200 Subject: [PATCH 4/5] fix: harden micro service callbacks --- src/Micro/Service.php | 42 +++++++++++-- tests/Integration/Micro/ServiceTest.php | 83 +++++++++++++++++++++++-- 2 files changed, 116 insertions(+), 9 deletions(-) diff --git a/src/Micro/Service.php b/src/Micro/Service.php index 0e5f9c7..e70fd05 100644 --- a/src/Micro/Service.php +++ b/src/Micro/Service.php @@ -40,6 +40,7 @@ public static function controlSubject(ServiceVerb $verb, string $name = '', stri private ?\Throwable $lastError = null; private int $handlerDepth = 0; private bool $pendingStopCleanup = false; + private bool $doneHandlerCalled = false; /** @var array */ private array $endpoints = []; @@ -272,6 +273,7 @@ public function stop(): void foreach ($subs as $sub) { $sub->unsubscribe(); } + $this->invokeDoneHandler(); } } else { $this->pendingStopCleanup = true; @@ -286,11 +288,17 @@ public function stop(): void 'Service stopped without graceful drain because the connection was not connected', )); } + + if ($this->handlerDepth > 0) { + $this->pendingStopCleanup = true; + } else { + $this->invokeDoneHandler(); + } + return; } - $doneHandler = $this->config->doneHandler; - if ($doneHandler !== null) { - $doneHandler($this); + if (!$this->pendingStopCleanup) { + $this->invokeDoneHandler(); } } @@ -419,7 +427,11 @@ private function recordError(\Throwable $e): void $errorHandler = $this->config->errorHandler; if ($errorHandler !== null) { - $errorHandler($this, $e); + try { + $errorHandler($this, $e); + } catch (\Throwable) { + // User error callbacks must not break service cleanup or reply flow. + } } } @@ -440,11 +452,13 @@ private function finishPendingStopCleanup(): void foreach ($subs as $sub) { $sub->unsubscribe(); } + $this->invokeDoneHandler(); return; } } $this->finalizeStoppedSubscriptions($subs); + $this->invokeDoneHandler(); } /** @@ -462,4 +476,24 @@ private function finalizeStoppedSubscriptions(array $subs): void } } } + + private function invokeDoneHandler(): void + { + if ($this->doneHandlerCalled) { + return; + } + + $this->doneHandlerCalled = true; + + $doneHandler = $this->config->doneHandler; + if ($doneHandler === null) { + return; + } + + try { + $doneHandler($this); + } catch (\Throwable $e) { + $this->recordError($e); + } + } } diff --git a/tests/Integration/Micro/ServiceTest.php b/tests/Integration/Micro/ServiceTest.php index 0b89938..c9a5fa2 100644 --- a/tests/Integration/Micro/ServiceTest.php +++ b/tests/Integration/Micro/ServiceTest.php @@ -264,6 +264,8 @@ public function testServiceReset(): void public function testServiceResetStartsNewStatsWindow(): void { + $subject = 'svc.reset.window.' . substr(uniqid(), -6); + $svc = Service::create($this->conn, new ServiceConfig( name: 'reset-window-svc', version: '1.0.0', @@ -271,9 +273,9 @@ public function testServiceResetStartsNewStatsWindow(): void $svc->addEndpoint('fail', function (): void { throw new \RuntimeException('boom'); - }, new EndpointConfig(subject: 'svc.reset.window')); + }, new EndpointConfig(subject: $subject)); - $reply = $this->conn->request('svc.reset.window', '', 2.0); + $reply = $this->conn->request($subject, '', 2.0); self::assertSame('boom', $reply->headers?->get('Nats-Service-Error')); self::assertSame('boom', $svc->lastError()?->getMessage()); @@ -324,6 +326,45 @@ public function testDoneHandler(): void self::assertTrue($doneCalled); } + public function testDoneHandlerRunsAfterDeferredStopCleanup(): void + { + $client = Connection::connect(self::natsUrl()); + $doneCalled = false; + $subscriptionsAtDone = null; + + try { + $svc = Service::create($this->conn, new ServiceConfig( + name: 'deferred-done-svc', + version: '1.0.0', + doneHandler: function () use (&$doneCalled, &$subscriptionsAtDone): void { + $doneCalled = true; + $subscriptionsAtDone = $this->conn->numSubscriptions(); + }, + )); + + $svc->addEndpoint('bye', function (RequestInterface $req) use ($svc): void { + $req->respond('bye'); + $svc->stop(); + }, new EndpointConfig(subject: 'svc.done.deferred')); + + $this->conn->flush(); + + $replyInbox = $client->newInbox(); + $replySub = $client->subscribeSync($replyInbox); + + $client->publish('svc.done.deferred', '', $replyInbox); + $client->flush(); + $this->conn->process(0.5); + $reply = $replySub->nextMessage(0.5); + + self::assertSame('bye', $reply->data); + self::assertTrue($doneCalled); + self::assertSame(0, $subscriptionsAtDone); + } finally { + $client->close(); + } + } + public function testStopIsIdempotent(): void { $svc = Service::create($this->conn, new ServiceConfig( @@ -412,6 +453,32 @@ public function testHandlerFailureIsExposedThroughLastErrorAndStats(): void $svc->stop(); } + public function testThrowingErrorHandlerDoesNotBreakErrorFlow(): void + { + $svc = Service::create($this->conn, new ServiceConfig( + name: 'throwing-error-handler-svc', + version: '1.0.0', + errorHandler: function (): void { + throw new \RuntimeException('error handler exploded'); + }, + )); + + $svc->addEndpoint('explode', function (): void { + throw new \RuntimeException('handler exploded'); + }, new EndpointConfig(subject: 'svc.throwing.error.handler')); + + $reply = $this->conn->request('svc.throwing.error.handler', '', 2.0); + + self::assertSame('handler exploded', $reply->headers?->get('Nats-Service-Error')); + self::assertSame('handler exploded', $svc->lastError()?->getMessage()); + + $stats = $svc->stats(); + self::assertSame(1, $stats->endpoints[0]->numErrors); + self::assertSame('handler exploded', $stats->endpoints[0]->lastError); + + $svc->stop(); + } + public function testRequestHasRespondedState(): void { $states = []; @@ -437,6 +504,8 @@ public function testRequestHasRespondedState(): void public function testResponseSendFailureIsObservable(): void { + $subject = 'svc.response.failure.' . substr(uniqid(), -6); + $svc = Service::create($this->conn, new ServiceConfig( name: 'response-failure-svc', version: '1.0.0', @@ -444,11 +513,15 @@ public function testResponseSendFailureIsObservable(): void $svc->addEndpoint('reply', function (RequestInterface $req): void { $req->respond('ok'); - }, new EndpointConfig(subject: 'svc.response.failure')); + }, new EndpointConfig(subject: $subject)); - $this->conn->publish('svc.response.failure', 'fire-and-forget'); + $this->conn->publish($subject, 'fire-and-forget'); $this->conn->flush(); - $this->conn->process(0.2); + + $deadline = microtime(true) + 1.0; + while ($svc->lastError() === null && microtime(true) < $deadline) { + $this->conn->process(0.05); + } self::assertSame('No reply subject', $svc->lastError()?->getMessage()); From 2b0b296b94eebff75019c8d02b8c2f8bbb569b51 Mon Sep 17 00:00:00 2001 From: Your Name Date: Mon, 30 Mar 2026 23:56:04 +0200 Subject: [PATCH 5/5] fix: harden micro reply handling --- src/Message.php | 9 +++++ src/Micro/Request.php | 35 ++++++++++++---- src/Micro/Service.php | 31 ++++++++------- tests/Integration/Micro/ServiceTest.php | 42 ++++++++++++++++++++ tests/Integration/Micro/StatsHandlerTest.php | 27 +++++++++++++ 5 files changed, 123 insertions(+), 21 deletions(-) diff --git a/src/Message.php b/src/Message.php index 49d845f..af52b7a 100644 --- a/src/Message.php +++ b/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,6 +33,9 @@ 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, @@ -38,6 +43,10 @@ public function respond(string $data, ?Headers $headers = null): void headers: $headers, ); $this->connection->publishMessage($msg); + + if ($this->afterRespond !== null) { + ($this->afterRespond)(); + } } public function hasHeaders(): bool diff --git a/src/Micro/Request.php b/src/Micro/Request.php index 97300b9..603bed6 100644 --- a/src/Micro/Request.php +++ b/src/Micro/Request.php @@ -12,11 +12,22 @@ final class Request implements RequestInterface { private bool $responded = false; + private readonly Message $msg; + private readonly Connection $connection; - public function __construct( - private readonly Message $msg, - private readonly Connection $connection, - ) {} + public function __construct(Message $msg, Connection $connection) + { + $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(...), + ); + } public function respond(string $data, ?Headers $headers = null): void { @@ -75,9 +86,7 @@ private function sendResponse(string $data, ?Headers $headers = null): void throw new NatsException('No reply subject'); } - if ($this->responded) { - throw new NatsException('Response already sent'); - } + $this->assertCanRespond(); $response = new Message( subject: $this->msg->replyTo, @@ -85,6 +94,18 @@ private function sendResponse(string $data, ?Headers $headers = null): void 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; } } diff --git a/src/Micro/Service.php b/src/Micro/Service.php index e70fd05..54e7a97 100644 --- a/src/Micro/Service.php +++ b/src/Micro/Service.php @@ -352,27 +352,29 @@ private function setupControlSubjects(): void private function handlePing(Message $msg): void { - $response = [ - 'type' => 'io.nats.micro.v1.ping_response', - 'name' => $this->config->name, - 'id' => $this->id, - 'version' => $this->config->version, - ]; - if ($this->config->metadata !== []) { - $response['metadata'] = $this->config->metadata; - } + $this->replyControl($msg, function (): array { + $response = [ + 'type' => 'io.nats.micro.v1.ping_response', + 'name' => $this->config->name, + 'id' => $this->id, + 'version' => $this->config->version, + ]; + if ($this->config->metadata !== []) { + $response['metadata'] = $this->config->metadata; + } - $this->replyControl($msg, $response); + return $response; + }); } private function handleInfo(Message $msg): void { - $this->replyControl($msg, $this->info()->toArray()); + $this->replyControl($msg, fn (): array => $this->info()->toArray()); } private function handleStats(Message $msg): void { - $this->replyControl($msg, $this->stats()->toArray()); + $this->replyControl($msg, fn (): array => $this->stats()->toArray()); } /** @@ -405,13 +407,14 @@ private function queueGroupFor(EndpointConfig $config): ?string } /** - * @param array $payload + * @param \Closure(): array $payloadFactory */ - private function replyControl(Message $msg, array $payload): void + private function replyControl(Message $msg, \Closure $payloadFactory): void { $this->handlerDepth++; try { + $payload = $payloadFactory(); $msg->respond(json_encode($payload, JSON_THROW_ON_ERROR)); } catch (\Throwable $e) { $this->recordError($e); diff --git a/tests/Integration/Micro/ServiceTest.php b/tests/Integration/Micro/ServiceTest.php index c9a5fa2..588211c 100644 --- a/tests/Integration/Micro/ServiceTest.php +++ b/tests/Integration/Micro/ServiceTest.php @@ -11,6 +11,7 @@ use Sofascore\Nats\Micro\Service; use Sofascore\Nats\Micro\ServiceConfig; use Sofascore\Nats\NatsException; +use Sofascore\Nats\TimeoutException; use Sofascore\Nats\Tests\Integration\NatsUrlTrait; use PHPUnit\Framework\TestCase; @@ -502,6 +503,47 @@ public function testRequestHasRespondedState(): void $svc->stop(); } + public function testMsgRespondPreventsFallbackErrorReply(): void + { + $client = Connection::connect(self::natsUrl()); + $subject = 'svc.msg.respond.' . substr(uniqid(), -6); + + try { + $svc = Service::create($this->conn, new ServiceConfig( + name: 'msg-respond-svc', + version: '1.0.0', + )); + + $svc->addEndpoint('reply', function (RequestInterface $req): void { + $req->msg()->respond('manual'); + throw new \RuntimeException('after manual respond'); + }, new EndpointConfig(subject: $subject)); + + $this->conn->flush(); + + $replyInbox = $client->newInbox(); + $replySub = $client->subscribeSync($replyInbox); + + $client->publish($subject, '', $replyInbox); + $client->flush(); + $this->conn->process(0.5); + $this->conn->flush(); + + $reply = $replySub->nextMessage(0.5); + self::assertSame('manual', $reply->data); + self::assertSame('after manual respond', $svc->lastError()?->getMessage()); + + try { + $replySub->nextMessage(0.1); + self::fail('Did not expect a fallback error reply'); + } catch (TimeoutException) { + self::assertTrue(true); + } + } finally { + $client->close(); + } + } + public function testResponseSendFailureIsObservable(): void { $subject = 'svc.response.failure.' . substr(uniqid(), -6); diff --git a/tests/Integration/Micro/StatsHandlerTest.php b/tests/Integration/Micro/StatsHandlerTest.php index 2c7de38..768b0c3 100644 --- a/tests/Integration/Micro/StatsHandlerTest.php +++ b/tests/Integration/Micro/StatsHandlerTest.php @@ -8,6 +8,7 @@ use Sofascore\Nats\Micro\EndpointConfig; use Sofascore\Nats\Micro\Service; use Sofascore\Nats\Micro\ServiceConfig; +use Sofascore\Nats\TimeoutException; use Sofascore\Nats\Tests\Integration\NatsUrlTrait; use PHPUnit\Framework\TestCase; @@ -81,4 +82,30 @@ public function testStatsWithoutHandlerHasNullData(): void $service->stop(); } + + public function testThrowingStatsHandlerIsRecordedInsteadOfEscaping(): void + { + $service = Service::create($this->conn, new ServiceConfig( + name: 'stats-throwing-handler', + version: '1.0.0', + statsHandler: function (): never { + throw new \RuntimeException('stats boom'); + }, + )); + + $service->addEndpoint('ping', function ($req): void { + $req->respond('pong'); + }); + + $this->conn->request('ping', '', 2.0); + + try { + $this->conn->request('$SRV.STATS.stats-throwing-handler', '', 0.2); + self::fail('Expected stats request to time out when stats handler throws'); + } catch (TimeoutException) { + self::assertSame('stats boom', $service->lastError()?->getMessage()); + } + + $service->stop(); + } }