From 33f53246916cb2996a89e40c5950d77224a85dd8 Mon Sep 17 00:00:00 2001 From: "matkomat-agent-2[bot]" Date: Mon, 30 Mar 2026 14:27:03 +0200 Subject: [PATCH 1/4] fix: make ordered consumers recover and track state --- src/JetStream/Consumer/ConsumeContext.php | 2 +- src/JetStream/Consumer/MessagePoller.php | 2 +- src/JetStream/Consumer/OrderedConsumer.php | 409 ++++++++++++++++-- .../Consumer/OrderedConsumerConfig.php | 3 +- .../JetStream/OrderedConsumerTest.php | 121 ++++++ 5 files changed, 509 insertions(+), 28 deletions(-) create mode 100644 tests/Integration/JetStream/OrderedConsumerTest.php diff --git a/src/JetStream/Consumer/ConsumeContext.php b/src/JetStream/Consumer/ConsumeContext.php index da3bdc5..1eed023 100644 --- a/src/JetStream/Consumer/ConsumeContext.php +++ b/src/JetStream/Consumer/ConsumeContext.php @@ -12,7 +12,7 @@ final class ConsumeContext /** @internal */ public function __construct( - private readonly Consumer $consumer, + private readonly ConsumerInterface $consumer, private readonly \Closure $handler, private readonly ConsumeLoopOptions $options, ) {} diff --git a/src/JetStream/Consumer/MessagePoller.php b/src/JetStream/Consumer/MessagePoller.php index deebd69..43b9438 100644 --- a/src/JetStream/Consumer/MessagePoller.php +++ b/src/JetStream/Consumer/MessagePoller.php @@ -21,7 +21,7 @@ final class MessagePoller implements \IteratorAggregate /** @internal */ public function __construct( - private readonly Consumer $consumer, + private readonly ConsumerInterface $consumer, private readonly MessagePollOptions $options, ) {} diff --git a/src/JetStream/Consumer/OrderedConsumer.php b/src/JetStream/Consumer/OrderedConsumer.php index 2527a8c..b93c0da 100644 --- a/src/JetStream/Consumer/OrderedConsumer.php +++ b/src/JetStream/Consumer/OrderedConsumer.php @@ -4,16 +4,23 @@ namespace Sofascore\Nats\JetStream\Consumer; +use Sofascore\Nats\Enum\AckPolicy; +use Sofascore\Nats\Enum\DeliveryPolicy; use Sofascore\Nats\Inbox; use Sofascore\Nats\Internal\TypeCast as T; +use Sofascore\Nats\JetStream\Error\JetStreamException; use Sofascore\Nats\JetStream\JetStreamContext; use Sofascore\Nats\JetStream\Message\JetStreamMessage; +use Sofascore\Nats\JetStream\Message\JetStreamMsg; +use Sofascore\Nats\Message; use Sofascore\Nats\NatsException; +use Sofascore\Nats\TimeoutException; final class OrderedConsumer implements ConsumerInterface { private ?Consumer $inner = null; - private int $cursor = 0; + private int $streamSequence = 0; + private int $consumerSequence = 0; /** @internal */ public function __construct( @@ -24,32 +31,60 @@ public function __construct( public function fetch(int $batch, ?FetchOptions $opts = null): MessageBatch { - return $this->getInner()->fetch($batch, $opts); + $opts ??= new FetchOptions(); + + return $this->runPullLoop( + limit: $batch, + maxBytes: null, + opts: $opts, + noWait: false, + ); } public function fetchBytes(int $maxBytes, ?FetchOptions $opts = null): MessageBatch { - return $this->getInner()->fetchBytes($maxBytes, $opts); + $opts ??= new FetchOptions(); + + return $this->runPullLoop( + limit: PHP_INT_MAX, + maxBytes: $maxBytes, + opts: $opts, + noWait: false, + ); } public function fetchNoWait(int $batch): MessageBatch { - return $this->getInner()->fetchNoWait($batch); + return $this->runPullLoop( + limit: $batch, + maxBytes: null, + opts: new FetchOptions(timeout: 1.0), + noWait: true, + ); } public function consumeLoop(\Closure $handler, ?ConsumeLoopOptions $opts = null): ConsumeContext { - return $this->getInner()->consumeLoop($handler, $opts); + $opts ??= new ConsumeLoopOptions(); + + return new ConsumeContext($this, $handler, $opts); } public function poll(?MessagePollOptions $opts = null): MessagePoller { - return $this->getInner()->poll($opts); + $opts ??= new MessagePollOptions(); + + return new MessagePoller($this, $opts); } public function next(float $timeout = 5.0): JetStreamMessage { - return $this->getInner()->next($timeout); + $batch = $this->fetch(1, new FetchOptions(timeout: $timeout)); + foreach ($batch->messages() as $message) { + return $message; + } + + throw new TimeoutException('No message available'); } public function info(): ConsumerInfo @@ -59,38 +94,48 @@ public function info(): ConsumerInfo public function cachedInfo(): ConsumerInfo { - return $this->info(); + return $this->getInner()->cachedInfo(); } private function getInner(): Consumer { if ($this->inner === null) { - $this->createOrderedConsumer(); - } - if ($this->inner === null) { - throw new NatsException('Failed to create ordered consumer'); + $this->inner = $this->createOrderedConsumer(); } + return $this->inner; } - private function createOrderedConsumer(): void + private function createOrderedConsumer(): Consumer { + $filterSubject = $this->config->filterSubject; + $filterSubjects = $this->config->filterSubjects; + if ( + $this->streamSequence === 0 + && $this->config->deliverPolicy === DeliveryPolicy::LastPerSubject + && $filterSubject === null + && $filterSubjects === [] + ) { + $filterSubjects = ['>']; + } + + $deliverPolicy = $this->streamSequence > 0 + ? DeliveryPolicy::ByStartSequence + : $this->config->deliverPolicy; + $consumerConfig = new ConsumerConfig( name: Inbox::nuid(), - deliverPolicy: $this->cursor > 0 - ? \Sofascore\Nats\Enum\DeliveryPolicy::ByStartSequence - : $this->config->deliverPolicy, - optStartSeq: $this->cursor > 0 ? $this->cursor + 1 : $this->config->optStartSeq, - optStartTime: $this->config->optStartTime, - ackPolicy: \Sofascore\Nats\Enum\AckPolicy::None, - maxDeliver: 1, - maxAckPending: 0, + deliverPolicy: $deliverPolicy, + optStartSeq: $this->resolveOptStartSeq($deliverPolicy), + optStartTime: $this->resolveOptStartTime($deliverPolicy), + ackPolicy: AckPolicy::None, + replicas: 1, headersOnly: $this->config->headersOnly, - filterSubject: $this->config->filterSubject, - filterSubjects: $this->config->filterSubjects, + filterSubject: $filterSubject, + filterSubjects: $filterSubjects, replayPolicy: $this->config->replayPolicy, memoryStorage: true, - inactiveThreshold: 30.0, + inactiveThreshold: $this->config->inactiveThreshold, ); $data = $this->js->apiRequest( @@ -98,10 +143,324 @@ private function createOrderedConsumer(): void ['stream_name' => $this->stream, 'config' => $consumerConfig->toArray()], ); - $this->inner = new Consumer( + return new Consumer( $this->js, $this->stream, ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])), ); } + + private function resolveOptStartSeq(DeliveryPolicy $deliverPolicy): ?int + { + if ($this->streamSequence > 0) { + return $this->streamSequence + 1; + } + + return $deliverPolicy === DeliveryPolicy::ByStartSequence + ? $this->config->optStartSeq + : null; + } + + private function resolveOptStartTime(DeliveryPolicy $deliverPolicy): ?\DateTimeImmutable + { + if ($this->streamSequence > 0) { + return null; + } + + return $deliverPolicy === DeliveryPolicy::ByStartTime + ? $this->config->optStartTime + : null; + } + + private function runPullLoop(int $limit, ?int $maxBytes, FetchOptions $opts, bool $noWait): MessageBatch + { + $batch = new MessageBatch(); + $deadline = microtime(true) + $opts->timeout; + $received = 0; + $totalBytes = 0; + + while ($received < $limit && ($maxBytes === null || $totalBytes < $maxBytes)) { + $remainingTimeout = $noWait + ? 1.0 + : max(0.001, $deadline - microtime(true)); + + if (!$noWait && $remainingTimeout <= 0.001 && $received > 0) { + break; + } + + $pullResult = $this->pull( + batch: $limit - $received, + maxBytes: $maxBytes === null ? null : max(1, $maxBytes - $totalBytes), + opts: new FetchOptions( + timeout: $remainingTimeout, + heartbeat: $opts->heartbeat, + minPending: $opts->minPending, + minAckPending: $opts->minAckPending, + ), + noWait: $noWait, + ); + + $needsReset = false; + foreach ($pullResult['messages'] as $message) { + if (!$this->acceptMessage($message)) { + $needsReset = true; + break; + } + + $batch->addMessage($message); + $received++; + $totalBytes += strlen($message->data()); + + if ($received >= $limit || ($maxBytes !== null && $totalBytes >= $maxBytes)) { + break; + } + } + + if ($received >= $limit || ($maxBytes !== null && $totalBytes >= $maxBytes)) { + break; + } + + if ($needsReset || $this->shouldRecover($pullResult['status'])) { + $this->recover($deadline); + continue; + } + + if ( + $received === 0 + && $pullResult['messages'] === [] + && !$this->currentConsumerExists() + ) { + $this->recover($deadline); + continue; + } + + if ($pullResult['status'] !== null || $pullResult['messages'] === []) { + break; + } + } + + $batch->setComplete(); + + return $batch; + } + + /** + * @return array{ + * messages: list, + * status: null|array{code: int, description: string} + * } + */ + private function pull(int $batch, ?int $maxBytes, FetchOptions $opts, bool $noWait): array + { + $inbox = Inbox::generate(); + $sub = $this->js->connection()->subscribeSync($inbox); + + try { + $requestData = []; + if ($maxBytes !== null) { + $requestData['max_bytes'] = $maxBytes; + } else { + $requestData['batch'] = $batch; + if ($opts->minPending !== null) { + $requestData['min_pending'] = $opts->minPending; + } + if ($opts->minAckPending !== null) { + $requestData['min_ack_pending'] = $opts->minAckPending; + } + } + + if ($noWait) { + $requestData['no_wait'] = true; + } else { + $requestData['expires'] = (int) ($opts->timeout * 1_000_000_000); + if ($opts->heartbeat !== null) { + $requestData['idle_heartbeat'] = (int) ($opts->heartbeat * 1_000_000_000); + } + } + + $consumer = $this->getInner(); + $this->js->connection()->publish( + "\$JS.API.CONSUMER.MSG.NEXT.{$this->stream}.{$consumer->name()}", + json_encode($requestData, JSON_THROW_ON_ERROR), + $inbox, + ); + + $messages = []; + $status = null; + $received = 0; + $totalBytes = 0; + $deadline = microtime(true) + $opts->timeout + 1.0; + + while ($received < $batch && ($maxBytes === null || $totalBytes < $maxBytes) && microtime(true) < $deadline) { + try { + $message = $sub->nextMessage(max(0.001, $deadline - microtime(true))); + } catch (TimeoutException) { + break; + } + + if ($this->isStatusMessage($message)) { + $status = $this->extractStatus($message); + if ($status['code'] === 100) { + $status = null; + continue; + } + break; + } + + $jsMessage = new JetStreamMsg($message, $this->js->connection()); + $messages[] = $jsMessage; + $received++; + $totalBytes += strlen($message->data); + } + + return [ + 'messages' => $messages, + 'status' => $status, + ]; + } finally { + $sub->unsubscribe(); + } + } + + private function isStatusMessage(Message $message): bool + { + if ($message->headers === null) { + return false; + } + + if ($message->data !== '') { + return false; + } + + return $message->headers->get('Status') !== null; + } + + /** + * @return array{code: int, description: string} + */ + private function extractStatus(Message $message): array + { + $statusRaw = trim((string) $message->headers?->get('Status')); + $description = trim((string) ($message->headers?->get('Description') ?? '')); + $statusCode = (int) explode(' ', $statusRaw)[0]; + + if ($description === '') { + $description = trim(substr($statusRaw, strlen((string) $statusCode))); + } + + return [ + 'code' => $statusCode, + 'description' => $description, + ]; + } + + private function acceptMessage(JetStreamMessage $message): bool + { + $metadata = $message->metadata(); + if ($metadata->consumerSequence !== $this->consumerSequence + 1) { + return false; + } + + $this->consumerSequence = $metadata->consumerSequence; + $this->streamSequence = $metadata->streamSequence; + + return true; + } + + /** + * @param array{code: int, description: string}|null $status + */ + private function shouldRecover(?array $status): bool + { + if ($status === null) { + return false; + } + + $description = strtolower($status['description']); + + if ($status['code'] === 503) { + return true; + } + + if ($status['code'] !== 409) { + return false; + } + + if (str_contains($description, 'consumer deleted')) { + return true; + } + + if (str_contains($description, 'leadership change')) { + return true; + } + + return str_contains($description, 'server shutdown'); + } + + private function currentConsumerExists(): bool + { + if ($this->inner === null) { + return false; + } + + try { + $this->inner->info(); + return true; + } catch (JetStreamException) { + return false; + } + } + + private function recover(float $deadline): void + { + $attempts = 0; + $maxResetAttempts = $this->config->maxResetAttempts; + if ($maxResetAttempts !== null && $maxResetAttempts <= 0) { + $maxResetAttempts = null; + } + $lastError = null; + + while ($maxResetAttempts === null || $attempts < $maxResetAttempts) { + $attempts++; + + try { + $this->recreateConsumer(); + return; + } catch (\Throwable $error) { + $lastError = $error; + + if (microtime(true) >= $deadline) { + break; + } + + usleep(min(50_000 * $attempts, 250_000)); + } + } + + if ($maxResetAttempts !== null && $attempts >= $maxResetAttempts) { + throw new NatsException( + sprintf('Failed to recreate ordered consumer after %d attempt(s)', $attempts), + 0, + $lastError, + ); + } + + throw new NatsException('Failed to recreate ordered consumer before the pull deadline elapsed', 0, $lastError); + } + + private function recreateConsumer(): void + { + $current = $this->inner; + $this->inner = null; + $this->consumerSequence = 0; + + if ($current !== null) { + try { + $this->js->deleteConsumer($this->stream, $current->name()); + } catch (\Throwable) { + } + } + + $this->inner = $this->createOrderedConsumer(); + } } diff --git a/src/JetStream/Consumer/OrderedConsumerConfig.php b/src/JetStream/Consumer/OrderedConsumerConfig.php index ae63acb..2019b87 100644 --- a/src/JetStream/Consumer/OrderedConsumerConfig.php +++ b/src/JetStream/Consumer/OrderedConsumerConfig.php @@ -17,7 +17,8 @@ public function __construct( public ?int $optStartSeq = null, public ?\DateTimeImmutable $optStartTime = null, public ReplayPolicy $replayPolicy = ReplayPolicy::Instant, + public float $inactiveThreshold = 300.0, public bool $headersOnly = false, - public ?float $maxResetAttempts = null, + public ?int $maxResetAttempts = null, ) {} } diff --git a/tests/Integration/JetStream/OrderedConsumerTest.php b/tests/Integration/JetStream/OrderedConsumerTest.php new file mode 100644 index 0000000..422e227 --- /dev/null +++ b/tests/Integration/JetStream/OrderedConsumerTest.php @@ -0,0 +1,121 @@ +conn = Connection::connect(self::natsUrl()); + $this->js = $this->conn->jetStream(); + try { $this->js->deleteStream('TEST_ORDERED_CONSUMER'); } catch (\Throwable) {} + + $this->stream = $this->js->createStream(new StreamConfig( + name: 'TEST_ORDERED_CONSUMER', + subjects: ['test.ordered.>'], + storage: StorageType::Memory, + )); + + for ($i = 1; $i <= 6; $i++) { + $this->js->publish('test.ordered.events', json_encode(['id' => $i], JSON_THROW_ON_ERROR)); + } + } + + protected function tearDown(): void + { + try { $this->js->deleteStream('TEST_ORDERED_CONSUMER'); } catch (\Throwable) {} + $this->conn->close(); + } + + public function testOrderedConsumerCarriesForwardCursorAcrossRecreation(): void + { + $consumer = $this->stream->orderedConsumer(new OrderedConsumerConfig( + deliverPolicy: DeliveryPolicy::All, + )); + + $first = $consumer->next(1.0); + $second = $consumer->next(1.0); + $initialName = $consumer->cachedInfo()->name; + + self::assertSame(1, $first->metadata()->streamSequence); + self::assertSame(2, $second->metadata()->streamSequence); + + $this->stream->deleteConsumer($initialName); + + $third = $consumer->next(1.0); + + self::assertSame(3, $third->metadata()->streamSequence); + self::assertSame(1, $third->metadata()->consumerSequence); + self::assertNotSame($initialName, $consumer->cachedInfo()->name); + } + + public function testOrderedConsumerRecreatesWhenSequenceTrackingDetectsLoss(): void + { + $consumer = $this->stream->orderedConsumer(new OrderedConsumerConfig( + deliverPolicy: DeliveryPolicy::All, + )); + + $first = $consumer->next(1.0); + $initialName = $consumer->cachedInfo()->name; + + self::assertSame(1, $first->metadata()->streamSequence); + + $property = new \ReflectionProperty($consumer, 'consumerSequence'); + $property->setValue($consumer, 99); + + $second = $consumer->next(1.0); + + self::assertSame(2, $second->metadata()->streamSequence); + self::assertSame(1, $second->metadata()->consumerSequence); + self::assertNotSame($initialName, $consumer->cachedInfo()->name); + } + + public function testOrderedConsumerHonorsConfiguredResetAttemptLimit(): void + { + $consumer = $this->stream->orderedConsumer(new OrderedConsumerConfig( + deliverPolicy: DeliveryPolicy::All, + maxResetAttempts: 2, + )); + + $first = $consumer->next(1.0); + self::assertSame(1, $first->metadata()->streamSequence); + + $this->js->deleteStream('TEST_ORDERED_CONSUMER'); + + $this->expectException(NatsException::class); + $this->expectExceptionMessage('Failed to recreate ordered consumer after 2 attempt(s)'); + + $consumer->next(0.3); + } + + public function testOrderedConsumerUsesConfiguredInactiveThreshold(): void + { + $consumer = $this->stream->orderedConsumer(new OrderedConsumerConfig( + deliverPolicy: DeliveryPolicy::All, + inactiveThreshold: 45.0, + )); + + $info = $consumer->info(); + + self::assertSame(45.0, $info->config->inactiveThreshold); + } +} From da44a2310cb59cbba2a65282499716005b6f96fc Mon Sep 17 00:00:00 2001 From: "matkomat-agent-2[bot]" Date: Mon, 30 Mar 2026 14:39:07 +0200 Subject: [PATCH 2/4] fix: validate ordered consumer reset attempts --- src/JetStream/Consumer/OrderedConsumer.php | 3 -- .../Consumer/OrderedConsumerConfig.php | 6 +++- .../Consumer/OrderedConsumerConfigTest.php | 34 +++++++++++++++++++ 3 files changed, 39 insertions(+), 4 deletions(-) create mode 100644 tests/Unit/JetStream/Consumer/OrderedConsumerConfigTest.php diff --git a/src/JetStream/Consumer/OrderedConsumer.php b/src/JetStream/Consumer/OrderedConsumer.php index b93c0da..9deb41f 100644 --- a/src/JetStream/Consumer/OrderedConsumer.php +++ b/src/JetStream/Consumer/OrderedConsumer.php @@ -415,9 +415,6 @@ private function recover(float $deadline): void { $attempts = 0; $maxResetAttempts = $this->config->maxResetAttempts; - if ($maxResetAttempts !== null && $maxResetAttempts <= 0) { - $maxResetAttempts = null; - } $lastError = null; while ($maxResetAttempts === null || $attempts < $maxResetAttempts) { diff --git a/src/JetStream/Consumer/OrderedConsumerConfig.php b/src/JetStream/Consumer/OrderedConsumerConfig.php index 2019b87..ea4a8b8 100644 --- a/src/JetStream/Consumer/OrderedConsumerConfig.php +++ b/src/JetStream/Consumer/OrderedConsumerConfig.php @@ -20,5 +20,9 @@ public function __construct( public float $inactiveThreshold = 300.0, public bool $headersOnly = false, public ?int $maxResetAttempts = null, - ) {} + ) { + if ($this->maxResetAttempts !== null && $this->maxResetAttempts <= 0) { + throw new \InvalidArgumentException('maxResetAttempts must be greater than 0'); + } + } } diff --git a/tests/Unit/JetStream/Consumer/OrderedConsumerConfigTest.php b/tests/Unit/JetStream/Consumer/OrderedConsumerConfigTest.php new file mode 100644 index 0000000..4411b40 --- /dev/null +++ b/tests/Unit/JetStream/Consumer/OrderedConsumerConfigTest.php @@ -0,0 +1,34 @@ +maxResetAttempts); + } + + public function testZeroMaxResetAttemptsIsRejected(): void + { + $this->expectException(\InvalidArgumentException::class); + $this->expectExceptionMessage('maxResetAttempts must be greater than 0'); + + new OrderedConsumerConfig(maxResetAttempts: 0); + } + + public function testNegativeMaxResetAttemptsIsRejected(): void + { + $this->expectException(\InvalidArgumentException::class); + $this->expectExceptionMessage('maxResetAttempts must be greater than 0'); + + new OrderedConsumerConfig(maxResetAttempts: -1); + } +} From ca176f82153837d7b61040808f62ff67d37ebf84 Mon Sep 17 00:00:00 2001 From: "matkomat-agent-2[bot]" Date: Mon, 30 Mar 2026 15:21:34 +0200 Subject: [PATCH 3/4] fix: preserve ordered consumer recovery semantics --- src/JetStream/Consumer/OrderedConsumer.php | 93 +++++++++++++------ .../JetStream/OrderedConsumerTest.php | 32 +++++++ 2 files changed, 96 insertions(+), 29 deletions(-) diff --git a/src/JetStream/Consumer/OrderedConsumer.php b/src/JetStream/Consumer/OrderedConsumer.php index 9deb41f..3dbe764 100644 --- a/src/JetStream/Consumer/OrderedConsumer.php +++ b/src/JetStream/Consumer/OrderedConsumer.php @@ -11,6 +11,7 @@ use Sofascore\Nats\JetStream\Error\JetStreamException; use Sofascore\Nats\JetStream\JetStreamContext; use Sofascore\Nats\JetStream\Message\JetStreamMessage; +use Sofascore\Nats\JetStream\Message\MessageMetadata; use Sofascore\Nats\JetStream\Message\JetStreamMsg; use Sofascore\Nats\Message; use Sofascore\Nats\NatsException; @@ -21,6 +22,8 @@ final class OrderedConsumer implements ConsumerInterface private ?Consumer $inner = null; private int $streamSequence = 0; private int $consumerSequence = 0; + /** @var array */ + private array $lastSubjectSequences = []; /** @internal */ public function __construct( @@ -100,34 +103,32 @@ public function cachedInfo(): ConsumerInfo private function getInner(): Consumer { if ($this->inner === null) { - $this->inner = $this->createOrderedConsumer(); + $this->inner = $this->createOrderedConsumer($this->streamSequence); + $this->consumerSequence = 0; } return $this->inner; } - private function createOrderedConsumer(): Consumer + private function createOrderedConsumer(int $resumeStreamSequence): Consumer { $filterSubject = $this->config->filterSubject; $filterSubjects = $this->config->filterSubjects; if ( - $this->streamSequence === 0 - && $this->config->deliverPolicy === DeliveryPolicy::LastPerSubject + $this->isLastPerSubject() && $filterSubject === null && $filterSubjects === [] ) { $filterSubjects = ['>']; } - $deliverPolicy = $this->streamSequence > 0 - ? DeliveryPolicy::ByStartSequence - : $this->config->deliverPolicy; + $deliverPolicy = $this->resolveDeliverPolicy($resumeStreamSequence); $consumerConfig = new ConsumerConfig( name: Inbox::nuid(), deliverPolicy: $deliverPolicy, - optStartSeq: $this->resolveOptStartSeq($deliverPolicy), - optStartTime: $this->resolveOptStartTime($deliverPolicy), + optStartSeq: $this->resolveOptStartSeq($resumeStreamSequence, $deliverPolicy), + optStartTime: $this->resolveOptStartTime($resumeStreamSequence, $deliverPolicy), ackPolicy: AckPolicy::None, replicas: 1, headersOnly: $this->config->headersOnly, @@ -150,10 +151,19 @@ private function createOrderedConsumer(): Consumer ); } - private function resolveOptStartSeq(DeliveryPolicy $deliverPolicy): ?int + private function resolveDeliverPolicy(int $resumeStreamSequence): DeliveryPolicy { - if ($this->streamSequence > 0) { - return $this->streamSequence + 1; + if ($resumeStreamSequence > 0 && !$this->isLastPerSubject()) { + return DeliveryPolicy::ByStartSequence; + } + + return $this->config->deliverPolicy; + } + + private function resolveOptStartSeq(int $resumeStreamSequence, DeliveryPolicy $deliverPolicy): ?int + { + if ($resumeStreamSequence > 0 && $deliverPolicy === DeliveryPolicy::ByStartSequence) { + return $resumeStreamSequence + 1; } return $deliverPolicy === DeliveryPolicy::ByStartSequence @@ -161,9 +171,9 @@ private function resolveOptStartSeq(DeliveryPolicy $deliverPolicy): ?int : null; } - private function resolveOptStartTime(DeliveryPolicy $deliverPolicy): ?\DateTimeImmutable + private function resolveOptStartTime(int $resumeStreamSequence, DeliveryPolicy $deliverPolicy): ?\DateTimeImmutable { - if ($this->streamSequence > 0) { + if ($resumeStreamSequence > 0) { return null; } @@ -178,6 +188,9 @@ private function runPullLoop(int $limit, ?int $maxBytes, FetchOptions $opts, boo $deadline = microtime(true) + $opts->timeout; $received = 0; $totalBytes = 0; + $tentativeStreamSequence = $this->streamSequence; + $tentativeConsumerSequence = $this->inner === null ? 0 : $this->consumerSequence; + $tentativeSubjectSequences = $this->lastSubjectSequences; while ($received < $limit && ($maxBytes === null || $totalBytes < $maxBytes)) { $remainingTimeout = $noWait @@ -202,14 +215,24 @@ private function runPullLoop(int $limit, ?int $maxBytes, FetchOptions $opts, boo $needsReset = false; foreach ($pullResult['messages'] as $message) { - if (!$this->acceptMessage($message)) { + $metadata = $message->metadata(); + if ($metadata->consumerSequence !== $tentativeConsumerSequence + 1) { $needsReset = true; break; } + $tentativeConsumerSequence = $metadata->consumerSequence; + if ($this->shouldSkipMessage($message, $metadata, $tentativeSubjectSequences)) { + continue; + } + $batch->addMessage($message); $received++; $totalBytes += strlen($message->data()); + $tentativeStreamSequence = $metadata->streamSequence; + if ($this->isLastPerSubject()) { + $tentativeSubjectSequences[$message->subject()] = $metadata->streamSequence; + } if ($received >= $limit || ($maxBytes !== null && $totalBytes >= $maxBytes)) { break; @@ -221,7 +244,8 @@ private function runPullLoop(int $limit, ?int $maxBytes, FetchOptions $opts, boo } if ($needsReset || $this->shouldRecover($pullResult['status'])) { - $this->recover($deadline); + $this->recover($deadline, $tentativeStreamSequence); + $tentativeConsumerSequence = 0; continue; } @@ -230,7 +254,8 @@ private function runPullLoop(int $limit, ?int $maxBytes, FetchOptions $opts, boo && $pullResult['messages'] === [] && !$this->currentConsumerExists() ) { - $this->recover($deadline); + $this->recover($deadline, $tentativeStreamSequence); + $tentativeConsumerSequence = 0; continue; } @@ -239,6 +264,9 @@ private function runPullLoop(int $limit, ?int $maxBytes, FetchOptions $opts, boo } } + $this->streamSequence = $tentativeStreamSequence; + $this->consumerSequence = $tentativeConsumerSequence; + $this->lastSubjectSequences = $tentativeSubjectSequences; $batch->setComplete(); return $batch; @@ -354,17 +382,24 @@ private function extractStatus(Message $message): array ]; } - private function acceptMessage(JetStreamMessage $message): bool - { - $metadata = $message->metadata(); - if ($metadata->consumerSequence !== $this->consumerSequence + 1) { + /** + * @param array $subjectSequences + */ + private function shouldSkipMessage( + JetStreamMessage $message, + MessageMetadata $metadata, + array $subjectSequences, + ): bool { + if (!$this->isLastPerSubject()) { return false; } - $this->consumerSequence = $metadata->consumerSequence; - $this->streamSequence = $metadata->streamSequence; + return $metadata->streamSequence <= ($subjectSequences[$message->subject()] ?? 0); + } - return true; + private function isLastPerSubject(): bool + { + return $this->config->deliverPolicy === DeliveryPolicy::LastPerSubject; } /** @@ -411,7 +446,7 @@ private function currentConsumerExists(): bool } } - private function recover(float $deadline): void + private function recover(float $deadline, int $resumeStreamSequence): void { $attempts = 0; $maxResetAttempts = $this->config->maxResetAttempts; @@ -421,7 +456,7 @@ private function recover(float $deadline): void $attempts++; try { - $this->recreateConsumer(); + $this->recreateConsumer($resumeStreamSequence); return; } catch (\Throwable $error) { $lastError = $error; @@ -445,11 +480,10 @@ private function recover(float $deadline): void throw new NatsException('Failed to recreate ordered consumer before the pull deadline elapsed', 0, $lastError); } - private function recreateConsumer(): void + private function recreateConsumer(int $resumeStreamSequence): void { $current = $this->inner; $this->inner = null; - $this->consumerSequence = 0; if ($current !== null) { try { @@ -458,6 +492,7 @@ private function recreateConsumer(): void } } - $this->inner = $this->createOrderedConsumer(); + $this->inner = $this->createOrderedConsumer($resumeStreamSequence); + $this->consumerSequence = 0; } } diff --git a/tests/Integration/JetStream/OrderedConsumerTest.php b/tests/Integration/JetStream/OrderedConsumerTest.php index 422e227..661590b 100644 --- a/tests/Integration/JetStream/OrderedConsumerTest.php +++ b/tests/Integration/JetStream/OrderedConsumerTest.php @@ -8,6 +8,7 @@ use Sofascore\Nats\Connection; use Sofascore\Nats\Enum\DeliveryPolicy; use Sofascore\Nats\Enum\StorageType; +use Sofascore\Nats\JetStream\Consumer\FetchOptions; use Sofascore\Nats\JetStream\Consumer\OrderedConsumerConfig; use Sofascore\Nats\JetStream\JetStreamContext; use Sofascore\Nats\JetStream\Stream\StreamConfig; @@ -89,6 +90,37 @@ public function testOrderedConsumerRecreatesWhenSequenceTrackingDetectsLoss(): v self::assertNotSame($initialName, $consumer->cachedInfo()->name); } + public function testOrderedConsumerPreservesLastPerSubjectAcrossRecreation(): void + { + $this->js->publish('test.ordered.latest.a', 'a1'); + $this->js->publish('test.ordered.latest.b', 'b1'); + $this->js->publish('test.ordered.latest.a', 'a2'); + $this->js->publish('test.ordered.latest.b', 'b2'); + $this->js->publish('test.ordered.latest.c', 'c1'); + $this->js->publish('test.ordered.latest.b', 'b3'); + + $consumer = $this->stream->orderedConsumer(new OrderedConsumerConfig( + deliverPolicy: DeliveryPolicy::LastPerSubject, + filterSubject: 'test.ordered.latest.>', + )); + + $first = $consumer->next(1.0); + $initialName = $consumer->cachedInfo()->name; + + self::assertSame('a2', $first->data()); + + $this->stream->deleteConsumer($initialName); + + $batch = $consumer->fetch(10, new FetchOptions(timeout: 1.0)); + $values = []; + foreach ($batch->messages() as $message) { + $values[] = $message->data(); + } + + self::assertSame(['c1', 'b3'], $values); + self::assertNotSame($initialName, $consumer->cachedInfo()->name); + } + public function testOrderedConsumerHonorsConfiguredResetAttemptLimit(): void { $consumer = $this->stream->orderedConsumer(new OrderedConsumerConfig( From 0db6ab23489e711041f76fa2edee2efd90128421 Mon Sep 17 00:00:00 2001 From: "matkomat-agent-2[bot]" Date: Mon, 30 Mar 2026 15:48:07 +0200 Subject: [PATCH 4/4] fix: recreate missing ordered consumer on info --- src/JetStream/Consumer/OrderedConsumer.php | 27 ++++++++++++++++++- .../JetStream/OrderedConsumerTest.php | 19 +++++++++++++ 2 files changed, 45 insertions(+), 1 deletion(-) diff --git a/src/JetStream/Consumer/OrderedConsumer.php b/src/JetStream/Consumer/OrderedConsumer.php index 3dbe764..878d88c 100644 --- a/src/JetStream/Consumer/OrderedConsumer.php +++ b/src/JetStream/Consumer/OrderedConsumer.php @@ -92,7 +92,17 @@ public function next(float $timeout = 5.0): JetStreamMessage public function info(): ConsumerInfo { - return $this->getInner()->info(); + try { + return $this->getInner()->info(); + } catch (JetStreamException $error) { + if (!$this->isMissingConsumerError($error)) { + throw $error; + } + + $this->recreateConsumer($this->streamSequence); + + return $this->getInner()->info(); + } } public function cachedInfo(): ConsumerInfo @@ -446,6 +456,21 @@ private function currentConsumerExists(): bool } } + private function isMissingConsumerError(JetStreamException $error): bool + { + $apiError = $error->apiError; + if ($apiError !== null && $apiError->errCode === 10014) { + return true; + } + + $description = strtolower($apiError !== null ? $apiError->description : $error->getMessage()); + $code = $apiError !== null ? $apiError->code : $error->getCode(); + + return $code === 404 + && str_contains($description, 'consumer') + && str_contains($description, 'not found'); + } + private function recover(float $deadline, int $resumeStreamSequence): void { $attempts = 0; diff --git a/tests/Integration/JetStream/OrderedConsumerTest.php b/tests/Integration/JetStream/OrderedConsumerTest.php index 661590b..0c76935 100644 --- a/tests/Integration/JetStream/OrderedConsumerTest.php +++ b/tests/Integration/JetStream/OrderedConsumerTest.php @@ -121,6 +121,25 @@ public function testOrderedConsumerPreservesLastPerSubjectAcrossRecreation(): vo self::assertNotSame($initialName, $consumer->cachedInfo()->name); } + public function testOrderedConsumerInfoRecoversAfterConsumerDeletion(): void + { + $consumer = $this->stream->orderedConsumer(new OrderedConsumerConfig( + deliverPolicy: DeliveryPolicy::All, + )); + + $first = $consumer->next(1.0); + $initialName = $consumer->cachedInfo()->name; + + self::assertSame(1, $first->metadata()->streamSequence); + + $this->stream->deleteConsumer($initialName); + + $info = $consumer->info(); + + self::assertNotSame($initialName, $info->name); + self::assertSame($info->name, $consumer->cachedInfo()->name); + } + public function testOrderedConsumerHonorsConfiguredResetAttemptLimit(): void { $consumer = $this->stream->orderedConsumer(new OrderedConsumerConfig(