fix: make consumer cachedInfo cache-only

src/JetStream/Consumer/Consumer.php

@@ -13,13 +13,12 @@
 
 final class Consumer implements ConsumerInterface
 {
-    private ?ConsumerInfo $cachedInfo = null;
-
     /** @internal */
     public function __construct(
         private readonly JetStreamContext $js,
         private readonly string $stream,
         private readonly ConsumerConfig $config,
+        private ?ConsumerInfo $cachedInfo = null,
     ) {}
 
     public function fetch(int $batch, ?FetchOptions $opts = null): MessageBatch
@@ -223,11 +222,8 @@ public function info(): ConsumerInfo
         return $this->cachedInfo;
     }
 
-    public function cachedInfo(): ConsumerInfo
+    public function cachedInfo(): ?ConsumerInfo
     {
-        if ($this->cachedInfo === null) {
-            return $this->info();
-        }
         return $this->cachedInfo;
     }
 

src/JetStream/Consumer/ConsumerInterface.php

@@ -29,5 +29,11 @@ public function next(float $timeout = 5.0): JetStreamMessage;
 
     public function info(): ConsumerInfo;
 
-    public function cachedInfo(): ConsumerInfo;
+    /**
+     * Returns the cached consumer info without making a server request.
+     *
+     * Handles created without an initial info snapshot return null until info()
+     * is called or the handle is otherwise hydrated.
+     */
+    public function cachedInfo(): ?ConsumerInfo;
 }

src/JetStream/Consumer/OrderedConsumer.php

@@ -105,9 +105,9 @@ public function info(): ConsumerInfo
         }
     }
 
-    public function cachedInfo(): ConsumerInfo
+    public function cachedInfo(): ?ConsumerInfo
     {
-        return $this->getInner()->cachedInfo();
+        return $this->inner?->cachedInfo();
     }
 
     private function getInner(): Consumer
@@ -158,6 +158,7 @@ private function createOrderedConsumer(int $resumeStreamSequence): Consumer
             $this->js,
             $this->stream,
             ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])),
+            ConsumerInfo::fromArray(T::stringKeyArray($data)),
         );
     }
 

src/JetStream/Consumer/PushConsumer.php

@@ -13,7 +13,6 @@
 
 final class PushConsumer implements PushConsumerInterface
 {
-    private ?ConsumerInfo $cachedInfo = null;
     private ?Subscription $subscription = null;
     private PushConsumerStatus $status = PushConsumerStatus::Unsubscribed;
     private ?PushConsumeContext $consumeContext = null;
@@ -27,6 +26,7 @@ public function __construct(
         private readonly JetStreamContext $js,
         private readonly string $stream,
         private readonly ConsumerConfig $config,
+        private ?ConsumerInfo $cachedInfo = null,
     ) {
         if ($this->config->deliverSubject === null) {
             throw new NatsException('Push consumer requires a deliver_subject in configuration');
@@ -95,11 +95,8 @@ public function info(): ConsumerInfo
         return $this->cachedInfo;
     }
 
-    public function cachedInfo(): ConsumerInfo
+    public function cachedInfo(): ?ConsumerInfo
     {
-        if ($this->cachedInfo === null) {
-            return $this->info();
-        }
         return $this->cachedInfo;
     }
 

src/JetStream/Consumer/PushConsumerInterface.php

@@ -14,9 +14,12 @@ interface PushConsumerInterface
     public function info(): ConsumerInfo;
 
     /**
-     * Returns the last cached consumer info without making a server request.
+     * Returns the cached consumer info without making a server request.
+     *
+     * Handles created without an initial info snapshot return null until info()
+     * is called or the handle is otherwise hydrated.
      */
-    public function cachedInfo(): ConsumerInfo;
+    public function cachedInfo(): ?ConsumerInfo;
 
     /**
      * Returns the consumer name.

src/JetStream/JetStreamContext.php

@@ -163,7 +163,12 @@ public function createConsumer(string $stream, ConsumerConfig $config): Consumer
             "CONSUMER.CREATE.{$stream}",
             ['stream_name' => $stream, 'config' => $config->toArray()],
         );
-        return new Consumer($this, $stream, ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])));
+        return new Consumer(
+            $this,
+            $stream,
+            ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])),
+            ConsumerInfo::fromArray(T::stringKeyArray($data)),
+        );
     }
 
     /**
@@ -175,7 +180,12 @@ public function createOrUpdateConsumer(string $stream, ConsumerConfig $config):
             ? "CONSUMER.CREATE.{$stream}.{$config->name}"
             : "CONSUMER.CREATE.{$stream}";
         $data = $this->apiRequest($op, ['stream_name' => $stream, 'config' => $config->toArray()]);
-        return new Consumer($this, $stream, ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])));
+        return new Consumer(
+            $this,
+            $stream,
+            ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])),
+            ConsumerInfo::fromArray(T::stringKeyArray($data)),
+        );
     }
 
     /**
@@ -184,7 +194,12 @@ public function createOrUpdateConsumer(string $stream, ConsumerConfig $config):
     public function consumer(string $stream, string $name): ConsumerInterface
     {
         $data = $this->apiRequest("CONSUMER.INFO.{$stream}.{$name}");
-        return new Consumer($this, $stream, ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])));
+        return new Consumer(
+            $this,
+            $stream,
+            ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])),
+            ConsumerInfo::fromArray(T::stringKeyArray($data)),
+        );
     }
 
     public function orderedConsumer(string $stream, ?OrderedConsumerConfig $config = null): ConsumerInterface
@@ -212,7 +227,12 @@ public function createPushConsumer(string $stream, ConsumerConfig $config): Push
             $op,
             ['stream_name' => $stream, 'config' => $config->toArray()],
         );
-        return new PushConsumer($this, $stream, ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])));
+        return new PushConsumer(
+            $this,
+            $stream,
+            ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])),
+            ConsumerInfo::fromArray(T::stringKeyArray($data)),
+        );
     }
 
     public function updatePushConsumer(string $stream, ConsumerConfig $config): PushConsumerInterface
@@ -225,7 +245,12 @@ public function updatePushConsumer(string $stream, ConsumerConfig $config): Push
             "CONSUMER.CREATE.{$stream}.{$consumerName}",
             ['stream_name' => $stream, 'config' => $config->toArray()],
         );
-        return new PushConsumer($this, $stream, ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])));
+        return new PushConsumer(
+            $this,
+            $stream,
+            ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])),
+            ConsumerInfo::fromArray(T::stringKeyArray($data)),
+        );
     }
 
     public function createOrUpdatePushConsumer(string $stream, ConsumerConfig $config): PushConsumerInterface
@@ -235,13 +260,23 @@ public function createOrUpdatePushConsumer(string $stream, ConsumerConfig $confi
             ? "CONSUMER.CREATE.{$stream}.{$consumerName}"
             : "CONSUMER.CREATE.{$stream}";
         $data = $this->apiRequest($op, ['stream_name' => $stream, 'config' => $config->toArray()]);
-        return new PushConsumer($this, $stream, ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])));
+        return new PushConsumer(
+            $this,
+            $stream,
+            ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])),
+            ConsumerInfo::fromArray(T::stringKeyArray($data)),
+        );
     }
 
     public function pushConsumer(string $stream, string $name): PushConsumerInterface
     {
         $data = $this->apiRequest("CONSUMER.INFO.{$stream}.{$name}");
-        return new PushConsumer($this, $stream, ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])));
+        return new PushConsumer(
+            $this,
+            $stream,
+            ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])),
+            ConsumerInfo::fromArray(T::stringKeyArray($data)),
+        );
     }
 
     // --- Publishing ---

src/JetStream/Stream/Stream.php

@@ -7,6 +7,7 @@
 use Sofascore\Nats\Internal\TypeCast as T;
 use Sofascore\Nats\JetStream\Consumer\Consumer;
 use Sofascore\Nats\JetStream\Consumer\ConsumerConfig;
+use Sofascore\Nats\JetStream\Consumer\ConsumerInfo;
 use Sofascore\Nats\JetStream\Consumer\ConsumerInterface;
 use Sofascore\Nats\JetStream\Consumer\ConsumerPauseResponse;
 use Sofascore\Nats\JetStream\Consumer\OrderedConsumer;
@@ -34,7 +35,12 @@ public function createConsumer(ConsumerConfig $config): ConsumerInterface
             "CONSUMER.CREATE.{$this->name}",
             ['stream_name' => $this->name, 'config' => $config->toArray()],
         );
-        return new Consumer($this->js, $this->name, ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])));
+        return new Consumer(
+            $this->js,
+            $this->name,
+            ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])),
+            ConsumerInfo::fromArray(T::stringKeyArray($data)),
+        );
     }
 
     public function createOrUpdateConsumer(ConsumerConfig $config): ConsumerInterface
@@ -47,7 +53,12 @@ public function createOrUpdateConsumer(ConsumerConfig $config): ConsumerInterfac
             $operation,
             ['stream_name' => $this->name, 'config' => $config->toArray()],
         );
-        return new Consumer($this->js, $this->name, ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])));
+        return new Consumer(
+            $this->js,
+            $this->name,
+            ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])),
+            ConsumerInfo::fromArray(T::stringKeyArray($data)),
+        );
     }
 
     public function updateConsumer(ConsumerConfig $config): ConsumerInterface
@@ -58,7 +69,12 @@ public function updateConsumer(ConsumerConfig $config): ConsumerInterface
     public function consumer(string $name): ConsumerInterface
     {
         $data = $this->js->apiRequest("CONSUMER.INFO.{$this->name}.{$name}");
-        return new Consumer($this->js, $this->name, ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])));
+        return new Consumer(
+            $this->js,
+            $this->name,
+            ConsumerConfig::fromArray(T::stringKeyArray($data['config'] ?? [])),
+            ConsumerInfo::fromArray(T::stringKeyArray($data)),
+        );
     }
 
     public function orderedConsumer(?OrderedConsumerConfig $config = null): ConsumerInterface

src/JetStream/Stream/StreamInterface.php

@@ -58,5 +58,10 @@ public function purge(?StreamPurgeOptions $options = null): int;
 
     public function info(bool $deletedDetails = false, ?string $subjectFilter = null): StreamInfo;
 
+    /**
+     * Returns the cached stream info without making a server request.
+     *
+     * Stream handles are created with an initial info snapshot already loaded.
+     */
     public function cachedInfo(): StreamInfo;
 }

tests/Integration/JetStream/ConsumerTest.php

@@ -9,10 +9,12 @@
 use Sofascore\Nats\Enum\AckPolicy;
 use Sofascore\Nats\Enum\DeliveryPolicy;
 use Sofascore\Nats\Enum\StorageType;
+use Sofascore\Nats\JetStream\Consumer\Consumer;
 use Sofascore\Nats\JetStream\Consumer\ConsumerConfig;
 use Sofascore\Nats\JetStream\Consumer\MessagePollOptions;
 use Sofascore\Nats\JetStream\Consumer\FetchOptions;
 use Sofascore\Nats\JetStream\Consumer\OrderedConsumerConfig;
+use Sofascore\Nats\JetStream\Error\JetStreamException;
 use Sofascore\Nats\JetStream\JetStreamContext;
 use Sofascore\Nats\JetStream\Stream\StreamConfig;
 use Sofascore\Nats\JetStream\Stream\StreamInterface;
@@ -60,6 +62,8 @@ public function testCreateConsumer(): void
             deliverPolicy: DeliveryPolicy::All,
         ));
 
+        self::assertSame('test-consumer-1', $consumer->cachedInfo()?->name);
+
         $info = $consumer->info();
         self::assertSame('TEST_CONSUMER', $info->stream);
         self::assertSame('test-consumer-1', $info->name);
@@ -331,6 +335,57 @@ public function testConsumerInfo(): void
         self::assertSame(0, $info->numAckPending);
     }
 
+    public function testCachedInfoHydratesColdConsumerHandleOnce(): void
+    {
+        $this->stream->createOrUpdateConsumer(new ConsumerConfig(
+            name: 'test-cached-cold',
+            durable: 'test-cached-cold',
+            ackPolicy: AckPolicy::Explicit,
+        ));
+
+        $consumer = new Consumer(
+            $this->js,
+            'TEST_CONSUMER',
+            new ConsumerConfig(
+                name: 'test-cached-cold',
+                durable: 'test-cached-cold',
+                ackPolicy: AckPolicy::Explicit,
+            ),
+        );
+
+        self::assertNull($consumer->cachedInfo());
+
+        $info = $consumer->info();
+        self::assertSame('test-cached-cold', $info->name);
+        self::assertSame('TEST_CONSUMER', $info->stream);
+
+        $this->stream->deleteConsumer('test-cached-cold');
+
+        self::assertSame('test-cached-cold', $consumer->cachedInfo()?->name);
+
+        $this->expectException(JetStreamException::class);
+        $consumer->info();
+    }
+
+    public function testStreamConsumerLookupSeedsCachedInfo(): void
+    {
+        $this->stream->createOrUpdateConsumer(new ConsumerConfig(
+            name: 'test-cached-lookup',
+            durable: 'test-cached-lookup',
+            ackPolicy: AckPolicy::Explicit,
+        ));
+
+        $consumer = $this->stream->consumer('test-cached-lookup');
+        self::assertSame('test-cached-lookup', $consumer->cachedInfo()?->name);
+
+        $this->stream->deleteConsumer('test-cached-lookup');
+
+        $info = $consumer->cachedInfo();
+        self::assertNotNull($info);
+        self::assertSame('test-cached-lookup', $info->name);
+        self::assertSame('TEST_CONSUMER', $info->stream);
+    }
+
     public function testListConsumers(): void
     {
         $this->stream->createOrUpdateConsumer(new ConsumerConfig(

tests/Integration/JetStream/OrderedConsumerTest.php

@@ -69,6 +69,31 @@ public function testOrderedConsumerCarriesForwardCursorAcrossRecreation(): void
         self::assertNotSame($initialName, $consumer->cachedInfo()->name);
     }
 
+    public function testOrderedConsumerCachedInfoIsCacheOnly(): void
+    {
+        $byFetch = $this->stream->orderedConsumer(new OrderedConsumerConfig(
+            deliverPolicy: DeliveryPolicy::All,
+        ));
+        self::assertNull($byFetch->cachedInfo());
+
+        $first = $byFetch->next(1.0);
+        self::assertSame(1, $first->metadata()->streamSequence);
+
+        $fetchInfo = $byFetch->cachedInfo();
+        self::assertNotNull($fetchInfo);
+        self::assertSame('TEST_ORDERED_CONSUMER', $fetchInfo->stream);
+        self::assertSame($fetchInfo->name, $first->metadata()->consumer);
+
+        $byInfo = $this->stream->orderedConsumer(new OrderedConsumerConfig(
+            deliverPolicy: DeliveryPolicy::All,
+        ));
+        self::assertNull($byInfo->cachedInfo());
+
+        $info = $byInfo->info();
+        self::assertSame($info->name, $byInfo->cachedInfo()?->name);
+        self::assertSame('TEST_ORDERED_CONSUMER', $info->stream);
+    }
+
     public function testOrderedConsumerRecreatesWhenSequenceTrackingDetectsLoss(): void
     {
         $consumer = $this->stream->orderedConsumer(new OrderedConsumerConfig(